canic-core 0.99.30

//!
//! Topology cascade workflow.
//!
//! Coordinates propagation of topology snapshots from root to leaves.
//! Enforces cascade invariants and delegates transport to `CascadeOps`.

use crate::{
    InternalError, InternalErrorOrigin,
    cdk::types::Principal,
    dto::cascade::TopologySnapshotInput,
    log,
    log::Topic,
    ops::{
        cascade::CascadeOps,
        ic::IcOps,
        runtime::{
            env::EnvOps,
            fleet_activation::FleetActivationRuntimeOps,
            metrics::cascade::{
                CascadeMetricOperation as MetricOperation, CascadeMetricOutcome as MetricOutcome,
                CascadeMetricReason as MetricReason, CascadeMetricSnapshot as MetricSnapshot,
                CascadeMetrics,
            },
        },
        storage::{children::CanisterChildrenOps, fleet_activation::FleetActivationOps},
    },
    workflow::{
        cascade::{
            snapshot::{
                TopologyDirectChild, TopologyPathNode, TopologySnapshot, TopologySnapshotBuilder,
                adapter::TopologySnapshotAdapter,
            },
            warn_if_large,
        },
        runtime::cycles::CycleWorkflow,
    },
};
use std::collections::HashMap;

///
/// TopologyCascadeWorkflow
/// Orchestrates topology snapshot propagation across the canister tree.
///
pub struct TopologyCascadeWorkflow;

fn prepared_topology_snapshot_hash(
    view: &TopologySnapshotInput,
) -> Result<Option<[u8; 32]>, InternalError> {
    if FleetActivationRuntimeOps::is_standalone_local() {
        return Ok(None);
    }
    crate::ops::fleet_activation::FleetActivationEvidenceOps::topology_snapshot_hash(view).map(Some)
}

fn prepared_topology_activation_evidence(
    activation_hash: Option<[u8; 32]>,
) -> Result<
    Option<crate::ops::storage::fleet_activation::PreparedFleetActivationSnapshot>,
    InternalError,
> {
    activation_hash
        .map(FleetActivationOps::prepare_applied_topology_snapshot)
        .transpose()
        .map_err(crate::ops::storage::StorageOpsError::from)
        .map_err(InternalError::from)
}

impl TopologyCascadeWorkflow {
    // ───────────────────────── Root cascades ─────────────────────────

    /// Initiates a topology cascade from the root canister toward `target_pid`.
    pub async fn root_cascade_topology_for_pid(
        target_pid: Principal,
    ) -> Result<TopologySnapshotInput, InternalError> {
        EnvOps::require_root()?;

        Self::record(
            MetricOperation::RootFanout,
            MetricOutcome::Started,
            MetricReason::Ok,
        );

        let snapshot = match TopologySnapshotBuilder::for_target(target_pid) {
            Ok(builder) => builder.build(),
            Err(err) => {
                Self::record(
                    MetricOperation::RootFanout,
                    MetricOutcome::Failed,
                    MetricReason::from_error(&err),
                );
                return Err(err);
            }
        };
        let target_input = Self::snapshot_input_for_target(target_pid, &snapshot)?;

        let root_pid = IcOps::canister_self();
        let first_child = match Self::next_child_on_path(root_pid, &snapshot.parents) {
            Ok(Some(first_child)) => first_child,
            Ok(None) => {
                Self::record(
                    MetricOperation::RouteResolve,
                    MetricOutcome::Skipped,
                    MetricReason::NoRoute,
                );
                Self::record(
                    MetricOperation::RootFanout,
                    MetricOutcome::Skipped,
                    MetricReason::NoRoute,
                );
                log!(
                    Topic::Sync,
                    Warn,
                    "sync.topology: no branch path to {target_pid}, skipping cascade"
                );
                return Ok(target_input);
            }
            Err(err) => {
                Self::record(
                    MetricOperation::RouteResolve,
                    MetricOutcome::Failed,
                    MetricReason::from_error(&err),
                );
                Self::record(
                    MetricOperation::RootFanout,
                    MetricOutcome::Failed,
                    MetricReason::from_error(&err),
                );
                return Err(err);
            }
        };

        let child_snapshot = match Self::slice_snapshot_for_child(first_child, &snapshot) {
            Ok(snapshot) => snapshot,
            Err(err) => {
                Self::record(
                    MetricOperation::RouteResolve,
                    MetricOutcome::Failed,
                    MetricReason::from_error(&err),
                );
                Self::record(
                    MetricOperation::RootFanout,
                    MetricOutcome::Failed,
                    MetricReason::from_error(&err),
                );
                return Err(err);
            }
        };
        Self::record(
            MetricOperation::RouteResolve,
            MetricOutcome::Completed,
            MetricReason::Ok,
        );

        match Self::send_snapshot(&first_child, &child_snapshot).await {
            Ok(()) => {
                Self::record(
                    MetricOperation::RootFanout,
                    MetricOutcome::Completed,
                    MetricReason::Ok,
                );
                Ok(target_input)
            }
            Err(err) => {
                Self::record(
                    MetricOperation::RootFanout,
                    MetricOutcome::Failed,
                    MetricReason::SendFailed,
                );
                Err(err)
            }
        }
    }

    pub(crate) fn root_snapshot_input_for_target(
        target_pid: Principal,
    ) -> Result<TopologySnapshotInput, InternalError> {
        EnvOps::require_root()?;
        let snapshot = TopologySnapshotBuilder::for_target(target_pid)?.build();
        Self::snapshot_input_for_target(target_pid, &snapshot)
    }

    fn snapshot_input_for_target(
        target_pid: Principal,
        snapshot: &TopologySnapshot,
    ) -> Result<TopologySnapshotInput, InternalError> {
        let target_snapshot = Self::slice_snapshot_for_child(target_pid, snapshot)?;
        Ok(TopologySnapshotAdapter::to_input(&target_snapshot))
    }

    // ──────────────────────── Non-root cascades ──────────────────────

    /// Continues a topology cascade on a non-root canister.
    pub async fn nonroot_cascade_topology(
        view: TopologySnapshotInput,
    ) -> Result<(), InternalError> {
        EnvOps::deny_root()?;
        let self_pid = IcOps::canister_self();
        CascadeOps::validate_topology_snapshot(
            &view,
            self_pid,
            EnvOps::parent_pid()?,
            &EnvOps::canister_role()?,
        )?;
        let activation_hash = prepared_topology_snapshot_hash(&view)?;
        let activation_evidence = prepared_topology_activation_evidence(activation_hash)?;

        let snapshot = TopologySnapshotAdapter::from_input(view);

        Self::record(
            MetricOperation::NonrootFanout,
            MetricOutcome::Started,
            MetricReason::Ok,
        );

        let next = match Self::next_child_on_path(self_pid, &snapshot.parents) {
            Ok(next) => next,
            Err(err) => {
                Self::record(
                    MetricOperation::RouteResolve,
                    MetricOutcome::Failed,
                    MetricReason::from_error(&err),
                );
                Self::record(
                    MetricOperation::NonrootFanout,
                    MetricOutcome::Failed,
                    MetricReason::from_error(&err),
                );
                return Err(err);
            }
        };

        let children = snapshot
            .children_map
            .get(&self_pid)
            .cloned()
            .unwrap_or_default();

        warn_if_large("nonroot fanout", children.len());

        Self::record(
            MetricOperation::LocalApply,
            MetricOutcome::Started,
            MetricReason::Ok,
        );

        Self::apply_local_topology(self_pid, children, activation_evidence);

        Self::record(
            MetricOperation::LocalApply,
            MetricOutcome::Completed,
            MetricReason::Ok,
        );

        CycleWorkflow::reconcile_after_topology_change()
            .map_err(|err| err.with_diagnostic_context("reconcile cycle top-up after topology"))?;

        if let Some(next_pid) = next {
            let next_snapshot = match Self::slice_snapshot_for_child(next_pid, &snapshot) {
                Ok(snapshot) => snapshot,
                Err(err) => {
                    Self::record(
                        MetricOperation::RouteResolve,
                        MetricOutcome::Failed,
                        MetricReason::from_error(&err),
                    );
                    Self::record(
                        MetricOperation::NonrootFanout,
                        MetricOutcome::Failed,
                        MetricReason::from_error(&err),
                    );
                    return Err(err);
                }
            };
            Self::record(
                MetricOperation::RouteResolve,
                MetricOutcome::Completed,
                MetricReason::Ok,
            );
            if let Err(err) = Self::send_snapshot(&next_pid, &next_snapshot).await {
                Self::record(
                    MetricOperation::NonrootFanout,
                    MetricOutcome::Failed,
                    MetricReason::SendFailed,
                );
                return Err(err);
            }
        } else {
            Self::record(
                MetricOperation::RouteResolve,
                MetricOutcome::Skipped,
                MetricReason::NoRoute,
            );
        }

        Self::record(
            MetricOperation::NonrootFanout,
            MetricOutcome::Completed,
            MetricReason::Ok,
        );

        Ok(())
    }

    // ───────────────────────── Internal helpers ──────────────────────

    fn apply_local_topology(
        self_pid: Principal,
        children: Vec<TopologyDirectChild>,
        activation_evidence: Option<
            crate::ops::storage::fleet_activation::PreparedFleetActivationSnapshot,
        >,
    ) {
        let entries = children
            .into_iter()
            .map(|child| (child.pid, child.role))
            .collect();
        CanisterChildrenOps::import_direct_children(self_pid, entries);
        if let Some(prepared) = activation_evidence {
            FleetActivationOps::commit_prepared_snapshot(prepared);
        }
    }

    // Record one topology cascade metric row using the fixed topology snapshot label.
    fn record(operation: MetricOperation, outcome: MetricOutcome, reason: MetricReason) {
        CascadeMetrics::record(operation, MetricSnapshot::Topology, outcome, reason);
    }

    // Send a topology snapshot to one child and record bounded transport outcome metrics.
    async fn send_snapshot(
        pid: &Principal,
        snapshot: &TopologySnapshot,
    ) -> Result<(), InternalError> {
        let view = TopologySnapshotAdapter::to_input(snapshot);

        Self::record(
            MetricOperation::ChildSend,
            MetricOutcome::Started,
            MetricReason::Ok,
        );

        match CascadeOps::send_topology_snapshot(*pid, &view).await {
            Ok(()) => {
                Self::record(
                    MetricOperation::ChildSend,
                    MetricOutcome::Completed,
                    MetricReason::Ok,
                );
                Ok(())
            }
            Err(err) => {
                Self::record(
                    MetricOperation::ChildSend,
                    MetricOutcome::Failed,
                    MetricReason::SendFailed,
                );
                Err(err
                    .with_diagnostic_context(format!("topology cascade rejected by child {pid}")))
            }
        }
    }

    // Resolve the next child hop from a topology parent chain rooted at this canister.
    fn next_child_on_path(
        self_pid: Principal,
        parents: &[TopologyPathNode],
    ) -> Result<Option<Principal>, InternalError> {
        let Some(first) = parents.first() else {
            return Err(InternalError::invariant(
                InternalErrorOrigin::Workflow,
                "topology parent chain is empty",
            ));
        };

        if first.pid != self_pid {
            return Err(InternalError::invariant(
                InternalErrorOrigin::Workflow,
                format!("topology parent chain does not start with self pid {self_pid}"),
            ));
        }

        Ok(parents.get(1).map(|p| p.pid))
    }

    // Slice a topology snapshot so the next child receives only its branch.
    fn slice_snapshot_for_child(
        next_pid: Principal,
        snapshot: &TopologySnapshot,
    ) -> Result<TopologySnapshot, InternalError> {
        let mut sliced_parents = Vec::new();
        let mut include = false;

        for parent in &snapshot.parents {
            if parent.pid == next_pid {
                include = true;
            }
            if include {
                sliced_parents.push(parent.clone());
            }
        }

        if sliced_parents.is_empty() {
            return Err(InternalError::invariant(
                InternalErrorOrigin::Workflow,
                format!("topology next hop {next_pid} not found in parent chain"),
            ));
        }

        let mut sliced_children_map = HashMap::new();
        for parent in &sliced_parents {
            let children = snapshot
                .children_map
                .get(&parent.pid)
                .cloned()
                .unwrap_or_default();
            sliced_children_map.insert(parent.pid, children);
        }

        Ok(TopologySnapshot {
            parents: sliced_parents,
            children_map: sliced_children_map,
        })
    }
}