canic-host 0.109.4

//! Module: install_root::fleet_component_provisioning_install
//!
//! Responsibility: drive the host-owned fresh-install Component transaction to terminal catalog publication.
//! Does not own: placement selection, Coordinator/root transitions, or catalog truth.
//! Boundary: passive status reconciles uncertain updates and the catalog is unreachable before exact `RuntimesActivated` evidence.

use super::{
    clock::current_unix_timestamp_secs,
    fleet_catalog_closeout::{
        PublishInstalledFleetCatalogRequest, publish_installed_fleet_catalog,
    },
    fleet_component_provisioning_journal::{
        FleetComponentProvisioningInstallPhase, PlanFleetComponentProvisioningInstallRequest,
        ResolvedFleetComponentProvisioningInstall, begin_component_provisioning_advance,
        begin_component_provisioning_preparation, begin_fleet_catalog_publication,
        complete_fleet_component_provisioning_install, plan_fleet_component_provisioning_install,
        record_component_provisioning_advanced, record_component_provisioning_observed,
        record_fleet_catalog_published,
    },
    fleet_component_provisioning_plan::{
        CompileFleetComponentProvisioningPlanRequest, compile_fleet_component_provisioning_plan,
    },
    icp_context::InstallIcpContext,
    operations::{call_with_arg, query_with_arg, resolve_install_protocol_binding},
    root_component_provisioning_operation_id,
};
use crate::{
    canister_protocol::CanisterProtocolError,
    fleet_catalog::FleetCatalogEntryV1,
    fleet_install_plan::PersistedFleetInstallPlan,
    icp::IcpCli,
    protocol_binding::ResolvedProtocolBinding,
    release_set::{
        AppConfigSnapshot, CanicInfrastructureRole,
        load_persisted_canic_infrastructure_artifact_manifest,
    },
};
use std::{path::Path, time::Duration};

use candid::Principal;
use canic_control_plane::dto::fleet_coordinator::{
    CoordinatorCommand, CoordinatorCommandResponse, CoordinatorOperationStatusResponse,
    CoordinatorStatusRequest, CoordinatorStatusResponse,
};
use canic_core::{
    cdk::utils::hash::hex_bytes,
    diagnostics::codes,
    dto::{
        component_provisioning::{
            FleetComponentProvisioningPhase, FleetComponentProvisioningStatusResponse,
        },
        fleet_registry::FleetRegistry,
        role::{OperationReceipt, OperationStatusRequest},
    },
    ids::FleetName,
    protocol,
};
use thiserror::Error as ThisError;

const BASE_ADVANCE_LIMIT: usize = 32;
const ADVANCES_PER_COMPONENT: usize = 8;
const ADVANCES_PER_PLACEMENT: usize = 4;
const ADVANCES_PER_ROOT: usize = 8;
const SCHEDULED_PROGRESS_POLL_INTERVAL: Duration = Duration::from_secs(1);

pub(super) struct InstallFleetComponentsRequest<'a> {
    pub icp: &'a InstallIcpContext,
    pub config_path: &'a Path,
    pub fleet_name: FleetName,
    pub fleet_install_plan: &'a PersistedFleetInstallPlan,
    pub coordinator: Principal,
    pub install_operation_id: [u8; 32],
    pub initial_active_registry: &'a FleetRegistry,
}

#[derive(Debug, ThisError)]
enum FleetComponentProvisioningInstallError {
    #[error(
        "Fleet Component provisioning exceeded its bounded advance count: phase={phase} accepted_roots={accepted_roots} acceptance_in_flight_root={acceptance_in_flight_root} operation_id={operation_id}"
    )]
    AdvanceBoundExceeded {
        phase: String,
        accepted_roots: String,
        acceptance_in_flight_root: String,
        operation_id: String,
    },

    #[error("Fleet Component provisioning could not calculate its bounded advance count")]
    AdvanceLimitOverflow,

    #[error("terminal Fleet Component provisioning has no published Fleet Registry version")]
    MissingPublishedRegistry,

    #[error("Fleet catalog publication has no exact durable row intent")]
    MissingCatalogIntent,

    #[error("Fleet Component provisioning status query failed after preparation: {0}")]
    StatusQuery(#[source] CanisterProtocolError),

    #[error("Fleet Component provisioning returned an unrelated command or status variant")]
    ResponseCorrelation,
}

/// Provision every explicitly placed initial Component, activate all roots, and publish discovery.
pub(super) fn install_fleet_components_and_publish_catalog(
    request: InstallFleetComponentsRequest<'_>,
) -> Result<(), Box<dyn std::error::Error>> {
    let config = AppConfigSnapshot::load(request.config_path)?;
    let infrastructure_manifest = load_persisted_canic_infrastructure_artifact_manifest(
        request.icp.root(),
        request.fleet_install_plan.plan.release_build_id,
    )?;
    let binding = resolve_install_protocol_binding(
        request.icp,
        &infrastructure_manifest,
        CanicInfrastructureRole::FleetCoordinator,
    )?;
    let compiled =
        compile_fleet_component_provisioning_plan(CompileFleetComponentProvisioningPlanRequest {
            config: config.model(),
            fleet_install_plan: &request.fleet_install_plan.plan,
            registry: request.initial_active_registry,
            operation_id: root_component_provisioning_operation_id(request.install_operation_id),
        })?;
    let advance_limit = provisioning_advance_limit(&compiled.prepare_request.plan)?;
    let mut current =
        plan_fleet_component_provisioning_install(PlanFleetComponentProvisioningInstallRequest {
            fleet_install_plan: request.fleet_install_plan,
            coordinator: request.coordinator,
            fleet_name: request.fleet_name.clone(),
            environment: request.icp.environment().to_string(),
            compiled,
        })?;
    let icp = request.icp.cli();
    let mut remote_advances = 0_usize;

    loop {
        current = match current.journal.phase {
            FleetComponentProvisioningInstallPhase::Planned => {
                begin_component_provisioning_preparation(&current)?
            }
            FleetComponentProvisioningInstallPhase::PreparationInFlight => {
                let status = query_or_prepare(icp, &binding, request.coordinator, &current)?;
                record_component_provisioning_observed(&current, status)?
            }
            FleetComponentProvisioningInstallPhase::Prepared => {
                begin_component_provisioning_advance(&current)?
            }
            FleetComponentProvisioningInstallPhase::AdvanceInFlight => {
                remote_advances = remote_advances
                    .checked_add(1)
                    .ok_or_else(|| advance_bound_exceeded(&current))?;
                if remote_advances > advance_limit {
                    return Err(advance_bound_exceeded(&current).into());
                }
                let status = reconcile_or_advance(icp, &binding, request.coordinator, &current)?;
                let poll_delay =
                    scheduled_progress_poll_delay(current.journal.last_status.as_ref(), &status);
                let advanced = record_component_provisioning_advanced(&current, status)?;
                if let Some(delay) = poll_delay {
                    std::thread::sleep(delay);
                }
                advanced
            }
            FleetComponentProvisioningInstallPhase::RuntimesActivated => {
                begin_fleet_catalog_publication(
                    &current,
                    catalog_entry(&request, current_unix_timestamp_secs()?),
                )?
            }
            FleetComponentProvisioningInstallPhase::CatalogPublicationInFlight => {
                let committed = publish_catalog(&request, &current)?;
                record_fleet_catalog_published(&current, committed)?
            }
            FleetComponentProvisioningInstallPhase::CatalogPublished => {
                complete_fleet_component_provisioning_install(&current)?
            }
            FleetComponentProvisioningInstallPhase::Complete => return Ok(()),
        };
    }
}

fn query_or_prepare(
    icp: &IcpCli,
    binding: &ResolvedProtocolBinding,
    coordinator: Principal,
    current: &ResolvedFleetComponentProvisioningInstall,
) -> Result<FleetComponentProvisioningStatusResponse, Box<dyn std::error::Error>> {
    match query_status_response(icp, binding, coordinator, current) {
        Ok(response) => correlate_status(response, current).map_err(Into::into),
        Err(error) if error.is_rejected_with(codes::STATE_UNAVAILABLE) => {
            let response: CoordinatorCommandResponse = call_with_arg(
                icp,
                binding,
                coordinator,
                protocol::CANIC_COMMAND,
                &CoordinatorCommand::ProvisionComponents(current.journal.prepare_request.clone()),
            )?;
            let CoordinatorCommandResponse::OperationAccepted(OperationReceipt { operation_id }) =
                response
            else {
                return Err(FleetComponentProvisioningInstallError::ResponseCorrelation.into());
            };
            if operation_id != current.journal.prepare_request.operation_id {
                return Err(FleetComponentProvisioningInstallError::ResponseCorrelation.into());
            }
            correlate_status(
                query_status_response(icp, binding, coordinator, current)?,
                current,
            )
            .map_err(Into::into)
        }
        Err(error) => Err(FleetComponentProvisioningInstallError::StatusQuery(error).into()),
    }
}

fn reconcile_or_advance(
    icp: &IcpCli,
    binding: &ResolvedProtocolBinding,
    coordinator: Principal,
    current: &ResolvedFleetComponentProvisioningInstall,
) -> Result<FleetComponentProvisioningStatusResponse, Box<dyn std::error::Error>> {
    correlate_status(
        query_status_response(icp, binding, coordinator, current)?,
        current,
    )
    .map_err(Into::into)
}

fn query_status_response(
    icp: &IcpCli,
    binding: &ResolvedProtocolBinding,
    coordinator: Principal,
    current: &ResolvedFleetComponentProvisioningInstall,
) -> Result<CoordinatorStatusResponse, CanisterProtocolError> {
    query_with_arg(
        icp,
        binding,
        coordinator,
        protocol::CANIC_STATUS,
        &CoordinatorStatusRequest::Operation(OperationStatusRequest {
            operation_id: current.journal.prepare_request.operation_id,
        }),
    )
}

fn correlate_status(
    response: CoordinatorStatusResponse,
    current: &ResolvedFleetComponentProvisioningInstall,
) -> Result<FleetComponentProvisioningStatusResponse, FleetComponentProvisioningInstallError> {
    let CoordinatorStatusResponse::Operation(
        CoordinatorOperationStatusResponse::ComponentProvisioning(status),
    ) = response
    else {
        return Err(FleetComponentProvisioningInstallError::ResponseCorrelation);
    };
    if status.plan_hash != current.journal.plan_hash {
        return Err(FleetComponentProvisioningInstallError::ResponseCorrelation);
    }
    Ok(status)
}

fn publish_catalog(
    request: &InstallFleetComponentsRequest<'_>,
    current: &ResolvedFleetComponentProvisioningInstall,
) -> Result<crate::fleet_catalog::CommittedFleetCatalog, Box<dyn std::error::Error>> {
    let status = current
        .journal
        .last_status
        .as_ref()
        .ok_or(FleetComponentProvisioningInstallError::MissingPublishedRegistry)?;
    if status.phase != FleetComponentProvisioningPhase::RuntimesActivated {
        return Err(FleetComponentProvisioningInstallError::MissingPublishedRegistry.into());
    }
    let terminal_fleet_registry = status
        .published_fleet_registry
        .as_ref()
        .ok_or(FleetComponentProvisioningInstallError::MissingPublishedRegistry)?;
    let catalog_entry = current
        .journal
        .catalog_entry
        .as_ref()
        .ok_or(FleetComponentProvisioningInstallError::MissingCatalogIntent)?;
    publish_installed_fleet_catalog(PublishInstalledFleetCatalogRequest {
        icp: request.icp,
        config_path: request.config_path,
        fleet_name: catalog_entry.fleet_name.clone(),
        fleet_install_plan: request.fleet_install_plan,
        coordinator: request.coordinator,
        deployed_at_unix_secs: catalog_entry.deployed_at_unix_secs,
        terminal_fleet_registry,
    })
}

fn catalog_entry(
    request: &InstallFleetComponentsRequest<'_>,
    deployed_at_unix_secs: u64,
) -> FleetCatalogEntryV1 {
    let fleet = &request.fleet_install_plan.plan.fleet;
    FleetCatalogEntryV1 {
        canonical_network_id: fleet.fleet.canonical_network_id,
        fleet_id: fleet.fleet.fleet_id,
        fleet_name: request.fleet_name.clone(),
        app: fleet.app.clone(),
        environment: request.icp.environment().to_string(),
        deployed_at_unix_secs,
        release_build_id: request.fleet_install_plan.plan.release_build_id,
        coordinator_principal: request.coordinator.to_text(),
    }
}

fn provisioning_advance_limit(
    plan: &canic_core::dto::component_provisioning::FleetComponentProvisioningPlan,
) -> Result<usize, FleetComponentProvisioningInstallError> {
    let placements = plan
        .batches
        .iter()
        .try_fold(0_usize, |total, batch| {
            total.checked_add(batch.placements.len())
        })
        .ok_or(FleetComponentProvisioningInstallError::AdvanceLimitOverflow)?;
    let components = plan
        .batches
        .iter()
        .flat_map(|batch| &batch.placements)
        .try_fold(0_usize, |total, placement| {
            total.checked_add(placement.entries.len())
        })
        .ok_or(FleetComponentProvisioningInstallError::AdvanceLimitOverflow)?;
    BASE_ADVANCE_LIMIT
        .checked_add(
            components
                .checked_mul(ADVANCES_PER_COMPONENT)
                .ok_or(FleetComponentProvisioningInstallError::AdvanceLimitOverflow)?,
        )
        .and_then(|limit| {
            placements
                .checked_mul(ADVANCES_PER_PLACEMENT)
                .and_then(|placement_work| limit.checked_add(placement_work))
        })
        .and_then(|limit| {
            plan.directory_confirmation_roots
                .len()
                .checked_mul(ADVANCES_PER_ROOT)
                .and_then(|root_work| limit.checked_add(root_work))
        })
        .ok_or(FleetComponentProvisioningInstallError::AdvanceLimitOverflow)
}

fn scheduled_progress_poll_delay(
    previous: Option<&FleetComponentProvisioningStatusResponse>,
    observed: &FleetComponentProvisioningStatusResponse,
) -> Option<Duration> {
    previous
        .is_some_and(|previous| previous == observed)
        .then_some(SCHEDULED_PROGRESS_POLL_INTERVAL)
}

fn advance_bound_exceeded(
    current: &ResolvedFleetComponentProvisioningInstall,
) -> FleetComponentProvisioningInstallError {
    let status = current.journal.last_status.as_ref();
    FleetComponentProvisioningInstallError::AdvanceBoundExceeded {
        phase: status.map_or_else(
            || "unavailable".to_string(),
            |status| format!("{:?}", status.phase),
        ),
        accepted_roots: status.map_or_else(
            || "unavailable".to_string(),
            |status| status.accepted_root_count.to_string(),
        ),
        acceptance_in_flight_root: status.map_or_else(
            || "unavailable".to_string(),
            |status| {
                status
                    .acceptance_in_flight_root
                    .as_ref()
                    .map_or_else(|| "none".to_string(), Principal::to_text)
            },
        ),
        operation_id: hex_bytes(current.journal.prepare_request.operation_id),
    }
}

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::{
        FleetComponentProvisioningInstallError, SCHEDULED_PROGRESS_POLL_INTERVAL,
        scheduled_progress_poll_delay,
    };
    use candid::Principal;
    use canic_core::{
        dto::component_provisioning::{
            FleetComponentProvisioningOperation, FleetComponentProvisioningPhase,
            FleetComponentProvisioningStatusResponse,
        },
        ids::{
            AppId, CanonicalNetworkId, ComponentDeploymentConfigurationDigest, FleetBinding,
            FleetCoordinatorBinding, FleetId, FleetKey, FleetRegistryAuthority, SubnetId,
        },
    };

    #[test]
    fn bounded_advance_error_renders_last_correlated_state() {
        let error = FleetComponentProvisioningInstallError::AdvanceBoundExceeded {
            phase: "AcceptingRoots".to_string(),
            accepted_roots: "0".to_string(),
            acceptance_in_flight_root: "aaaaa-aa".to_string(),
            operation_id: "07".repeat(32),
        };

        assert_eq!(
            error.to_string(),
            format!(
                "Fleet Component provisioning exceeded its bounded advance count: phase=AcceptingRoots accepted_roots=0 acceptance_in_flight_root=aaaaa-aa operation_id={}",
                "07".repeat(32)
            )
        );
    }

    #[test]
    fn unchanged_scheduled_status_waits_without_delaying_observed_progress() {
        let previous = status(FleetComponentProvisioningPhase::AcceptingRoots);
        assert_eq!(
            scheduled_progress_poll_delay(Some(&previous), &previous),
            Some(SCHEDULED_PROGRESS_POLL_INTERVAL)
        );
        let advanced = status(FleetComponentProvisioningPhase::RootsAccepted);
        assert_eq!(
            scheduled_progress_poll_delay(Some(&previous), &advanced),
            None
        );
        assert_eq!(scheduled_progress_poll_delay(None, &previous), None);
    }

    fn status(phase: FleetComponentProvisioningPhase) -> FleetComponentProvisioningStatusResponse {
        let fleet = FleetBinding {
            fleet: FleetKey {
                canonical_network_id: CanonicalNetworkId::ic_mainnet(),
                fleet_id: FleetId::from_generated_bytes([1; 32]),
            },
            app: AppId::from("demo"),
        };
        FleetComponentProvisioningStatusResponse {
            operation_id: [2; 32],
            plan_hash: [3; 32],
            fleet_registry: canic_core::dto::fleet_registry::FleetRegistryVersion {
                authority: FleetRegistryAuthority {
                    binding: FleetCoordinatorBinding {
                        fleet,
                        coordinator_subnet: SubnetId::from_principal(Principal::from_slice(
                            &[4; 29],
                        )),
                        coordinator: Principal::from_slice(&[5; 29]),
                    },
                    epoch: 1,
                },
                revision: 3,
                content_hash: [6; 32],
            },
            configuration_digest: ComponentDeploymentConfigurationDigest::from_bytes([7; 32]),
            operation: FleetComponentProvisioningOperation::FreshInstall,
            phase,
            directory_confirmation_root_count: 1,
            root_batch_count: 1,
            accepted_root_count: 0,
            acceptance_in_flight_root: None,
            provisioned_root_count: 0,
            current_root: None,
            provisioning_in_flight_root: None,
            directory_confirmed_root_count: 0,
            current_synchronization: None,
            current_publication: None,
            publication_in_flight_root: None,
            runtime_activated_root_count: 0,
            current_activation: None,
            activation_in_flight_root: None,
            pending_root_failure: None,
            group_placement_count: 5,
            component_count: 5,
            planned_at_ns: 1,
            roots_accepted_at_ns: None,
            components_provisioned_at_ns: None,
            published_fleet_registry: None,
            service_topology_published_at_ns: None,
            directories_confirmed_at_ns: None,
            runtimes_activated_at_ns: None,
        }
    }
}