staircase 0.0.7

use std::time::Duration;

use k8s_openapi::{NamespaceResourceScope, serde::Serialize};
use kube::{Resource, core::object::HasStatus, runtime::controller::Action};

#[cfg(feature = "controller_metrics")]
use opentelemetry::{
    global,
    metrics::{Counter, Histogram, Meter},
};

use super::{MutableStepOutcome, RunContext, RunError, RunPlan, StatusUpdater, status::status_log_and_ignore};

#[derive(Clone, Debug)]
/// Result of one reconcile run.
///
/// Use [`RunOutcome::action`] to convert this into a recommended Kubernetes
/// controller [`Action`].
///
/// [`Action`]: kube::runtime::controller::Action
pub enum RunOutcome {
    /// The run stopped without making a modification.
    ///
    /// This usually means the controller is waiting for external state to
    /// change or become observable.
    Stopped,
    /// A step made one meaningful modification.
    ///
    /// Immediate requeue is recommended so the next run can observe the new
    /// state and continue reconciliation.
    Modified,
    /// No step made a modification.
    ///
    /// Current state appears to match desired state for this plan.
    NoOp,
}

impl RunOutcome {
    /// Generates a recommended controller action for this outcome.
    pub fn action(&self, default_reconcile: Duration) -> Action {
        match self {
            RunOutcome::Stopped | RunOutcome::NoOp => Action::requeue(default_reconcile),
            RunOutcome::Modified => Action::requeue(Duration::from_secs(0)),
        }
    }
}

#[derive(Clone, Debug)]
pub struct RunStepConfig {
    #[cfg(feature = "controller_metrics")]
    pub(crate) step_duration: Histogram<u64>,
    #[cfg(feature = "controller_metrics")]
    pub(crate) status_errors: Counter<u64>,
}

#[allow(clippy::derivable_impls)]
impl Default for RunStepConfig {
    fn default() -> Self {
        #[cfg(feature = "controller_metrics")]
        return Self::with_meter(global::meter("reconciler"));
        #[cfg(not(feature = "controller_metrics"))]
        return Self {};
    }
}

impl RunStepConfig {
    #[cfg(feature = "controller_metrics")]
    pub fn with_meter(meter: Meter) -> Self {
        Self {
            step_duration: meter
                .u64_histogram("step.duration")
                .with_unit("ms")
                .with_description("runtime of a single reconcile step")
                .with_boundaries(vec![
                    0.0, 1.0, 2.0, 4.0, 8.0, 15.0, 20.0, 40.0, 75.0, 125.0, 250.0, 800.0, 2500.0,
                ])
                .build(),
            status_errors: meter
                .u64_counter("step.status.error.count")
                .with_description("count of all errors during updating the status while reconciling")
                .build(),
        }
    }
}

/// Executes a plan with an already constructed [`RunContext`].
///
/// Most users should call [`crate::controller::run`] through a
/// [`crate::controller::Reconciler`].
/// Use this function when you need custom context construction, status updates,
/// or execution configuration.
pub async fn run_plan<R, P, SU>(
    context: RunContext<R>,
    data: P::InData,
    plan: P,
    status_updater: &SU,
    #[allow(unused_variables)] config: &RunStepConfig,
) -> Result<RunOutcome, RunError<P::Error>>
where
    R: Resource<Scope = NamespaceResourceScope, DynamicType: Default> + HasStatus,
    R::Status: Serialize + Send,
    P: RunPlan<Resource = R>,
    SU: StatusUpdater<R>,
{
    #[cfg(feature = "controller_trace")]
    {
        let meta = context.resource.meta();
        tracing::debug!(
            start_time = ?context.run_start,
            resource = meta.name.clone().unwrap_or_default(),
            namespace = meta.namespace.clone().unwrap_or_default(),
            version = meta.resource_version.clone().unwrap_or_default(),
            "start run_plan"
        );
    }

    match plan.run_plan(&context, data, config).await? {
        MutableStepOutcome::Stop => Ok(RunOutcome::Stopped),
        MutableStepOutcome::Modified { status } => {
            if let Some(status) = status
                && let Err(e) = status_updater.update(&context, status).await
            {
                status_log_and_ignore(e);

                #[cfg(feature = "controller_metrics")]
                config.status_errors.add(1, &[]);
            }
            Ok(RunOutcome::Modified)
        },
        MutableStepOutcome::NoModification { .. } => Ok(RunOutcome::NoOp),
    }
}

#[cfg(test)]
pub(crate) mod test_support {
    use super::*;

    use std::{
        convert::Infallible,
        error::Error,
        fmt,
        future::Future,
        sync::{Arc, Mutex},
    };

    use k8s_openapi::{apimachinery::pkg::apis::meta::v1::ObjectMeta, jiff::Timestamp};
    use kube::{CustomResource, client::Body};
    use schemars::JsonSchema;
    use serde::{Deserialize, Serialize};

    use crate::controller::{Immutable, Mutable, MutableStepOutcome, MutableStepResult, Step, StepOutcome, StepResult};

    #[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
    #[kube(
        group = "tests.staircase.dev",
        version = "v1",
        kind = "RunTest",
        status = "RunTestStatus",
        namespaced
    )]
    pub struct RunTestSpec {}

    #[derive(Deserialize, Serialize, Clone, Debug, Default, PartialEq, Eq, JsonSchema)]
    pub struct RunTestStatus {
        pub observed: i32,
    }

    #[derive(Debug, PartialEq, Eq)]
    pub enum TestError {
        Step(&'static str),
        Status,
    }

    impl fmt::Display for TestError {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            match self {
                TestError::Step(name) => write!(f, "step {name} failed"),
                TestError::Status => f.write_str("status failed"),
            }
        }
    }

    impl Error for TestError {}

    pub struct FakeStatusUpdater {
        pub fail:     bool,
        pub statuses: Arc<Mutex<Vec<RunTestStatus>>>,
    }

    impl StatusUpdater<RunTest> for FakeStatusUpdater {
        type Error = TestError;

        fn update(
            &self,
            _context: &RunContext<RunTest>,
            status: RunTestStatus,
        ) -> impl Future<Output = Result<(), Self::Error>> + Send {
            let fail = self.fail;
            let statuses = self.statuses.clone();
            async move {
                statuses.lock().unwrap().push(status);
                if fail { Err(TestError::Status) } else { Ok(()) }
            }
        }
    }

    pub enum FakeStepOutcome<Out> {
        NoModification { data: Out },
        Stop,
        Error,
    }

    pub struct FakeStep<In, Out> {
        pub name:    &'static str,
        pub outcome: FakeStepOutcome<Out>,
        pub log:     Arc<Mutex<Vec<&'static str>>>,
        pub _in:     std::marker::PhantomData<In>,
    }

    impl<In, Out> FakeStep<In, Out> {
        pub fn no_modification(name: &'static str, data: Out, log: Arc<Mutex<Vec<&'static str>>>) -> Self {
            Self {
                name,
                outcome: FakeStepOutcome::NoModification { data },
                log,
                _in: std::marker::PhantomData,
            }
        }

        pub fn stop(name: &'static str, log: Arc<Mutex<Vec<&'static str>>>) -> Self {
            Self {
                name,
                outcome: FakeStepOutcome::Stop,
                log,
                _in: std::marker::PhantomData,
            }
        }

        pub fn error(name: &'static str, log: Arc<Mutex<Vec<&'static str>>>) -> Self {
            Self {
                name,
                outcome: FakeStepOutcome::Error,
                log,
                _in: std::marker::PhantomData,
            }
        }
    }

    impl<In, Out> Step for FakeStep<In, Out>
    where
        In: Send + Sync + 'static,
        Out: Clone + Send + Sync + 'static,
    {
        type Error = TestError;
        type InData = In;
        type Mode = Immutable;
        type OutData = Out;
        type Resource = RunTest;

        async fn run(
            &self,
            _context: &RunContext<Self::Resource>,
            _data: Self::InData,
        ) -> StepResult<Self::OutData, Self::Error> {
            self.log.lock().unwrap().push(self.name);
            match &self.outcome {
                FakeStepOutcome::NoModification { data } => Ok(StepOutcome::NoModification { data: data.clone() }),
                FakeStepOutcome::Stop => Ok(StepOutcome::Stop),
                FakeStepOutcome::Error => Err(TestError::Step(self.name)),
            }
        }

        fn traced_name(&self) -> std::borrow::Cow<'static, str> { std::borrow::Cow::Borrowed(self.name) }
    }

    pub enum FakeMutableStepOutcome<Out> {
        NoModification { data: Out },
        Modified { status: Option<RunTestStatus> },
    }

    pub struct FakeMutableStep<In, Out> {
        pub name:    &'static str,
        pub outcome: FakeMutableStepOutcome<Out>,
        pub log:     Arc<Mutex<Vec<&'static str>>>,
        pub _in:     std::marker::PhantomData<In>,
    }

    impl<In, Out> FakeMutableStep<In, Out> {
        pub fn no_modification(name: &'static str, data: Out, log: Arc<Mutex<Vec<&'static str>>>) -> Self {
            Self {
                name,
                outcome: FakeMutableStepOutcome::NoModification { data },
                log,
                _in: std::marker::PhantomData,
            }
        }

        pub fn modified(name: &'static str, status: Option<RunTestStatus>, log: Arc<Mutex<Vec<&'static str>>>) -> Self {
            Self {
                name,
                outcome: FakeMutableStepOutcome::Modified { status },
                log,
                _in: std::marker::PhantomData,
            }
        }
    }

    impl<In, Out> Step for FakeMutableStep<In, Out>
    where
        In: Send + Sync + 'static,
        Out: Clone + Send + Sync + 'static,
    {
        type Error = TestError;
        type InData = In;
        type Mode = Mutable;
        type OutData = Out;
        type Resource = RunTest;

        async fn run(
            &self,
            _context: &RunContext<Self::Resource>,
            _data: Self::InData,
        ) -> MutableStepResult<Self::Resource, Self::OutData, Self::Error> {
            self.log.lock().unwrap().push(self.name);
            match &self.outcome {
                FakeMutableStepOutcome::NoModification { data } => {
                    Ok(MutableStepOutcome::NoModification { data: data.clone() })
                },
                FakeMutableStepOutcome::Modified { status } => {
                    Ok(MutableStepOutcome::Modified { status: status.clone() })
                },
            }
        }

        fn traced_name(&self) -> std::borrow::Cow<'static, str> { std::borrow::Cow::Borrowed(self.name) }
    }

    pub struct NumberToStringStep {
        pub seen: Arc<Mutex<Vec<i32>>>,
    }

    impl Step for NumberToStringStep {
        type Error = TestError;
        type InData = i32;
        type Mode = Immutable;
        type OutData = String;
        type Resource = RunTest;

        async fn run(
            &self,
            _context: &RunContext<Self::Resource>,
            data: Self::InData,
        ) -> StepResult<Self::OutData, Self::Error> {
            self.seen.lock().unwrap().push(data);
            Ok(StepOutcome::NoModification {
                data: format!("value-{data}"),
            })
        }
    }

    pub struct StringRecordingStep {
        pub seen: Arc<Mutex<Vec<String>>>,
    }

    impl Step for StringRecordingStep {
        type Error = TestError;
        type InData = String;
        type Mode = Immutable;
        type OutData = ();
        type Resource = RunTest;

        async fn run(
            &self,
            _context: &RunContext<Self::Resource>,
            data: Self::InData,
        ) -> StepResult<Self::OutData, Self::Error> {
            self.seen.lock().unwrap().push(data);
            Ok(StepOutcome::NoModification { data: () })
        }
    }

    pub fn context() -> RunContext<RunTest> {
        let resource = RunTest {
            metadata: ObjectMeta {
                name: Some("sample".to_string()),
                namespace: Some("apps".to_string()),
                ..Default::default()
            },
            spec:     RunTestSpec {},
            status:   None,
        };
        let service = tower::service_fn(|_request: http::Request<Body>| async {
            Ok::<_, Infallible>(http::Response::new(Body::empty()))
        });
        RunContext {
            resource:  Arc::new(resource),
            client:    kube::Client::new(service, "default"),
            run_start: Timestamp::now(),
        }
    }

    pub fn status_updater(fail: bool) -> (FakeStatusUpdater, Arc<Mutex<Vec<RunTestStatus>>>) {
        let statuses = Arc::new(Mutex::new(Vec::new()));
        (
            FakeStatusUpdater {
                fail,
                statuses: statuses.clone(),
            },
            statuses,
        )
    }
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{Arc, Mutex},
        time::Duration,
    };

    use kube::runtime::controller::Action;

    use crate::controller::{Plan, RunError, StepFailure, StepOutcome, immutable_step, mutable_step};

    use super::*;
    use test_support::*;

    #[test]
    fn run_outcome_action_maps_modified_to_immediate_requeue() {
        let default_reconcile = Duration::from_secs(30);

        assert_eq!(
            RunOutcome::Stopped.action(default_reconcile),
            Action::requeue(default_reconcile)
        );
        assert_eq!(
            RunOutcome::NoOp.action(default_reconcile),
            Action::requeue(default_reconcile)
        );
        assert_eq!(
            RunOutcome::Modified.action(default_reconcile),
            Action::requeue(Duration::from_secs(0))
        );
    }

    #[test]
    fn typed_steps_run_in_order_and_return_noop() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeStep::no_modification("first", 1_i32, log.clone()))
            .then(FakeStep::no_modification("second", "loaded".to_string(), log.clone()))
            .then(FakeStep::no_modification("third", (), log.clone()));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::NoOp));
        assert_eq!(*log.lock().unwrap(), vec!["first", "second", "third"]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn typed_step_can_pass_number_to_step_that_outputs_string() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let seen_numbers = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeStep::no_modification("number", 7_i32, log.clone()))
            .then(NumberToStringStep {
                seen: seen_numbers.clone(),
            })
            .then(FakeStep::no_modification("string", (), log.clone()));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::NoOp));
        assert_eq!(*seen_numbers.lock().unwrap(), vec![7]);
        assert_eq!(*log.lock().unwrap(), vec!["number", "string"]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn mutable_step_can_continue_to_next_step() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let seen = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeMutableStep::no_modification(
            "mutable",
            "loaded".to_string(),
            log.clone(),
        ))
        .then(StringRecordingStep { seen: seen.clone() });

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::NoOp));
        assert_eq!(*log.lock().unwrap(), vec!["mutable"]);
        assert_eq!(*seen.lock().unwrap(), vec!["loaded".to_string()]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn immutable_closure_step_can_continue_to_next_step() {
        let seen = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(immutable_step(
            "closure-load",
            |context: RunContext<RunTest>, data: i32| async move {
                let name = context.resource.metadata.name.clone().unwrap_or_default();
                Ok::<_, TestError>(StepOutcome::NoModification {
                    data: format!("{name}-{data}"),
                })
            },
        ))
        .then(StringRecordingStep { seen: seen.clone() });

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), 7_i32, plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::NoOp));
        assert_eq!(*seen.lock().unwrap(), vec!["sample-7".to_string()]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn mutable_closure_step_can_modify_with_status() {
        let (status_updater, statuses) = status_updater(false);
        let status = RunTestStatus { observed: 7 };
        let plan = Plan::from(mutable_step::<RunTest, (), (), TestError, _, _>(
            "closure-modify",
            move |_context: RunContext<RunTest>, _data: ()| {
                let status = status.clone();
                async move { Ok::<_, TestError>(MutableStepOutcome::Modified { status: Some(status) }) }
            },
        ));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::Modified));
        assert_eq!(*statuses.lock().unwrap(), vec![RunTestStatus { observed: 7 }]);
    }

    #[test]
    fn stop_skips_later_steps() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeStep::no_modification("first", 1_i32, log.clone()))
            .then(FakeStep::<i32, String>::stop("stop", log.clone()))
            .then(FakeStep::no_modification("skipped", (), log.clone()));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::Stopped));
        assert_eq!(*log.lock().unwrap(), vec!["first", "stop"]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn modified_without_status_skips_later_steps() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeMutableStep::<(), i32>::modified("modified", None, log.clone()))
            .then(FakeStep::no_modification("skipped", (), log.clone()));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::Modified));
        assert_eq!(*log.lock().unwrap(), vec!["modified"]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn modified_with_status_updates_status_once() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let status = RunTestStatus { observed: 42 };
        let plan = Plan::from(FakeMutableStep::<(), i32>::modified(
            "modified",
            Some(status.clone()),
            log.clone(),
        ));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::Modified));
        assert_eq!(*statuses.lock().unwrap(), vec![status]);
    }

    #[test]
    fn status_update_failure_is_ignored_after_modification() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(true);
        let status = RunTestStatus { observed: 42 };
        let plan = Plan::from(FakeMutableStep::<(), i32>::modified(
            "modified",
            Some(status.clone()),
            log.clone(),
        ));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::Modified));
        assert_eq!(*statuses.lock().unwrap(), vec![status]);
    }

    #[test]
    fn step_error_is_returned_and_skips_later_steps() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeStep::<(), i32>::error("error", log.clone())).then(FakeStep::no_modification(
            "skipped",
            (),
            log.clone(),
        ));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let error = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap_err();

        assert!(matches!(error, RunError::Step {
            source: TestError::Step("error"),
            ..
        }));
        assert_eq!(*log.lock().unwrap(), vec!["error"]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn run_error_source_exposes_underlying_step_error() {
        let error = RunError::from(StepFailure {
            step:   std::borrow::Cow::Borrowed("load"),
            source: TestError::Step("load"),
        });

        assert_eq!(error.to_string(), "step `load` failed");
        assert_eq!(
            std::error::Error::source(&error).map(ToString::to_string),
            Some("step load failed".to_string())
        );
    }

    #[test]
    fn join_success_returns_noop_and_runs_all_branches() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeStep::no_modification("load", 7_i32, log.clone()))
            .join((
                FakeStep::no_modification("left", "left".to_string(), log.clone()),
                FakeStep::no_modification("right", 99_i32, log.clone()),
            ))
            .then(FakeStep::no_modification("merge", (), log.clone()));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::NoOp));
        assert_eq!(*log.lock().unwrap(), vec!["load", "left", "right", "merge"]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn join_map_transforms_tuple_output_before_next_step() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let seen = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeStep::no_modification("load", 7_i32, log.clone()))
            .join_map(
                (
                    FakeStep::no_modification("left", "left".to_string(), log.clone()),
                    FakeStep::no_modification("right", 99_i32, log.clone()),
                ),
                |(left, right)| format!("{left}-{right}"),
            )
            .then(StringRecordingStep { seen: seen.clone() });

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::NoOp));
        assert_eq!(*log.lock().unwrap(), vec!["load", "left", "right"]);
        assert_eq!(*seen.lock().unwrap(), vec!["left-99".to_string()]);
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn join_returns_all_failures_in_tuple_order() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeStep::no_modification("load", 7_i32, log.clone())).join((
            FakeStep::<i32, String>::error("left", log.clone()),
            FakeStep::<i32, i32>::error("right", log.clone()),
        ));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let error = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap_err();

        let RunError::Parallel { errors } = error else {
            panic!("expected parallel error");
        };
        assert_eq!(errors.len(), 2);
        assert_eq!(errors[0].step, "left");
        assert_eq!(errors[1].step, "right");
        assert!(statuses.lock().unwrap().is_empty());
    }

    #[test]
    fn join_stop_returns_stopped_without_running_later_steps() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, _) = status_updater(false);
        let plan = Plan::from(FakeStep::no_modification("load", 7_i32, log.clone()))
            .join((
                FakeStep::<i32, String>::stop("left", log.clone()),
                FakeStep::no_modification("right", 99_i32, log.clone()),
            ))
            .then(FakeStep::no_modification("skipped", (), log.clone()));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let outcome = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap();

        assert!(matches!(outcome, RunOutcome::Stopped));
        assert_eq!(*log.lock().unwrap(), vec!["load", "left", "right"]);
    }

    #[test]
    fn join_error_takes_precedence_over_stop() {
        let log = Arc::new(Mutex::new(Vec::new()));
        let (status_updater, statuses) = status_updater(false);
        let plan = Plan::from(FakeStep::no_modification("load", 7_i32, log.clone())).join((
            FakeStep::<i32, String>::stop("left", log.clone()),
            FakeStep::<i32, i32>::error("right", log.clone()),
        ));

        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
        let error = rt
            .block_on(async { run_plan(context(), (), plan, &status_updater, &RunStepConfig::default()).await })
            .unwrap_err();

        let RunError::Parallel { errors } = error else {
            panic!("expected parallel error");
        };
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0].step, "right");
        assert_eq!(*log.lock().unwrap(), vec!["load", "left", "right"]);
        assert!(statuses.lock().unwrap().is_empty());
    }
}