camel_core/lifecycle/adapters/

controller_actor.rs

//! Actor loop and supervision — the async task that processes route control commands.
//!
//! Extracted from a monolithic file. The command enum and handle live
//! in [`controller_actor_commands`](super::controller_actor_commands).

use std::collections::HashSet;
use std::sync::{Arc, Mutex};
use std::time::Instant;

use camel_api::{CamelError, MetricsCollector, RouteController, SupervisionConfig};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};

pub(crate) use super::controller_actor_commands::RouteControllerCommand;
pub use super::controller_actor_commands::RouteControllerHandle;
use super::route_controller::DefaultRouteController;
use super::route_helpers::CrashNotification;

pub fn spawn_controller_actor(
    controller: DefaultRouteController,
) -> (RouteControllerHandle, tokio::task::JoinHandle<()>) {
    let (tx, mut rx) = mpsc::channel::<RouteControllerCommand>(256);
    // Clone the cohort gate into the handle BEFORE the controller moves
    // into the actor task. Reset/activate bypass the actor entirely (they
    // act on the shared gate), so the handle needs its own Arc.
    let cohort = Arc::clone(&controller.cohort);
    // Hold a clone of tx so the spawned task can send StartRoute back through
    // the same channel after the 100ms restart sleep, without moving the
    // original tx (which we still need to return as RouteControllerHandle).
    let tx_for_spawn = tx.clone();
    let handle = tokio::spawn(async move {
        let mut controller = controller;
        // Tracks routes currently restarting via spawned off-actor tasks.
        // Related to but separate from spawn_supervision_task's
        // currently_restarting set (that set tracks crash-recovery restarts;
        // this one tracks command-driven restarts).
        let restarting: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
        while let Some(cmd) = rx.recv().await {
            match cmd {
                RouteControllerCommand::StartRoute { route_id, reply } => {
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    if restarting
                        .lock()
                        .expect("restarting mutex poisoned") // allow-unwrap
                        .contains(&route_id)
                    {
                        let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                            "route {} is restarting",
                            route_id
                        ))));
                        continue;
                    }
                    let _ = reply.send(controller.start_route(&route_id).await);
                }
                RouteControllerCommand::StopRoute { route_id, reply } => {
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    if restarting
                        .lock()
                        .expect("restarting mutex poisoned") // allow-unwrap
                        .contains(&route_id)
                    {
                        let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                            "route {} is restarting",
                            route_id
                        ))));
                        continue;
                    }
                    let _ = reply.send(controller.stop_route(&route_id).await);
                }
                RouteControllerCommand::RestartRoute { route_id, reply } => {
                    // Reject if already restarting.
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    {
                        let mut guard = restarting.lock().expect("restarting mutex poisoned"); // allow-unwrap
                        if guard.contains(&route_id) {
                            let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                                "route {} is restarting",
                                route_id
                            ))));
                            continue;
                        }
                        guard.insert(route_id.clone());
                    }

                    // Stop inline — actor owns the controller, so we cannot
                    // move it into a spawned task. The 100ms sleep + start
                    // happen off-actor by sending a StartRoute back through
                    // the same channel.
                    let stop_result = controller.stop_route(&route_id).await;
                    if let Err(ref e) = stop_result {
                        // allow-unwrap: Mutex cannot be poisoned in normal operation
                        restarting
                            .lock()
                            .expect("restarting mutex poisoned") // allow-unwrap
                            .remove(&route_id);
                        let _ = reply.send(Err(e.clone()));
                        continue;
                    }

                    // Spawn only the sleep + send StartRoute back through
                    // the SAME channel. The StartRoute is processed on the
                    // actor thread (correct ownership), and the reply moves
                    // into the spawned task to bridge the restart caller.
                    let tx_clone = tx_for_spawn.clone();
                    let restarting_clone = restarting.clone();
                    let route_id_for_start = route_id.clone();
                    let route_id_for_cleanup = route_id.clone();
                    tokio::spawn(async move {
                        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                        // Remove from `restarting` BEFORE sending StartRoute: the
                        // actor's StartRoute arm rejects if the route is in the
                        // restarting set, so removing first lets the actor
                        // accept its own self-sent command.
                        // allow-unwrap: Mutex cannot be poisoned in normal operation
                        restarting_clone
                            .lock()
                            .expect("restarting mutex poisoned") // allow-unwrap
                            .remove(&route_id_for_cleanup);
                        if tx_clone
                            .send(RouteControllerCommand::StartRoute {
                                route_id: route_id_for_start,
                                reply,
                            })
                            .await
                            .is_err()
                        {
                            warn!(
                                "route {} restart: StartRoute send failed (actor shutting down)",
                                route_id_for_cleanup
                            );
                        }
                    });
                    // Actor returns to rx.recv() immediately — HoL eliminated.
                }
                RouteControllerCommand::SuspendRoute { route_id, reply } => {
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    if restarting
                        .lock()
                        .expect("restarting mutex poisoned") // allow-unwrap
                        .contains(&route_id)
                    {
                        let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                            "route {} is restarting",
                            route_id
                        ))));
                        continue;
                    }
                    let _ = reply.send(controller.suspend_route(&route_id).await);
                }
                RouteControllerCommand::ResumeRoute { route_id, reply } => {
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    if restarting
                        .lock()
                        .expect("restarting mutex poisoned") // allow-unwrap
                        .contains(&route_id)
                    {
                        let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                            "route {} is restarting",
                            route_id
                        ))));
                        continue;
                    }
                    let _ = reply.send(controller.resume_route(&route_id).await);
                }
                RouteControllerCommand::StartAllRoutes { reply } => {
                    let _ = reply.send(controller.start_all_routes().await);
                }
                RouteControllerCommand::StopAllRoutes { reply } => {
                    let _ = reply.send(controller.stop_all_routes().await);
                }
                RouteControllerCommand::AddRoute { definition, reply } => {
                    let _ = reply.send(controller.add_route(definition).await);
                }
                RouteControllerCommand::RemoveRoute { route_id, reply } => {
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    if restarting
                        .lock()
                        .expect("restarting mutex poisoned") // allow-unwrap
                        .contains(&route_id)
                    {
                        let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                            "route {} is restarting",
                            route_id
                        ))));
                        continue;
                    }
                    let _ = reply.send(controller.remove_route(&route_id).await);
                }
                RouteControllerCommand::SwapPipeline {
                    route_id,
                    pipeline,
                    reply,
                } => {
                    let _ = reply.send(controller.swap_pipeline(&route_id, pipeline));
                }
                RouteControllerCommand::SwapPipelineRaw {
                    route_id,
                    pipeline,
                    lifecycle,
                    reply,
                } => {
                    let _ =
                        reply.send(controller.swap_pipeline_raw(&route_id, pipeline, lifecycle));
                }
                RouteControllerCommand::CompileRouteDefinition { definition, reply } => {
                    let _ = reply.send(controller.compile_route_definition(definition));
                }
                RouteControllerCommand::CompileRouteDefinitionWithGeneration {
                    definition,
                    generation,
                    reply,
                } => {
                    let _ = reply.send(
                        controller.compile_route_definition_with_generation(definition, generation),
                    );
                }
                RouteControllerCommand::CompileRouteDefinitionPipeline {
                    definition,
                    generation,
                    reply,
                } => {
                    let _ = reply
                        .send(controller.compile_route_definition_pipeline(definition, generation));
                }
                RouteControllerCommand::CompileRouteDefinitionDryPipeline { definition, reply } => {
                    let _ =
                        reply.send(controller.compile_route_definition_dry_pipeline(definition));
                }
                RouteControllerCommand::PrepareRouteDefinitionWithGeneration {
                    definition,
                    generation,
                    reply,
                } => {
                    let _ = reply.send(
                        controller.prepare_route_definition_with_generation(definition, generation),
                    );
                }
                RouteControllerCommand::InsertPreparedRoute { prepared, reply } => {
                    let _ = reply.send(controller.insert_prepared_route(prepared));
                }
                RouteControllerCommand::DiscardPreparedStaging { route_id, reply } => {
                    controller.discard_prepared_staging(&route_id);
                    let _ = reply.send(Ok(()));
                }
                RouteControllerCommand::RemoveRoutePreservingFunctions { route_id, reply } => {
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    if restarting
                        .lock()
                        .expect("restarting mutex poisoned") // allow-unwrap
                        .contains(&route_id)
                    {
                        let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                            "route {} is restarting",
                            route_id
                        ))));
                        continue;
                    }
                    let _ = reply.send(
                        controller
                            .remove_route_preserving_functions(&route_id)
                            .await,
                    );
                }
                RouteControllerCommand::RouteFromUri { route_id, reply } => {
                    let _ = reply.send(controller.route_from_uri(&route_id));
                }
                RouteControllerCommand::SetErrorHandler { config } => {
                    controller.set_error_handler(config);
                }
                RouteControllerCommand::SetTracerConfig { config } => {
                    controller.set_tracer_config(&config);
                }
                RouteControllerCommand::SetBindExposureAcks { acks } => {
                    controller.set_bind_exposure_acks(acks);
                }
                RouteControllerCommand::RouteCount { reply } => {
                    let _ = reply.send(controller.route_count());
                }
                RouteControllerCommand::InFlightCount { route_id, reply } => {
                    let _ = reply.send(controller.in_flight_count(&route_id));
                }
                RouteControllerCommand::RouteExists { route_id, reply } => {
                    let _ = reply.send(controller.route_exists(&route_id));
                }
                RouteControllerCommand::RouteIds { reply } => {
                    let _ = reply.send(controller.route_ids());
                }
                RouteControllerCommand::ListEndpoints { reply } => {
                    let _ = reply.send(controller.list_endpoint_uris());
                }
                RouteControllerCommand::RoutesForEndpoint { uri, reply } => {
                    let _ = reply.send(controller.routes_for_endpoint(&uri));
                }
                RouteControllerCommand::HealthCheckEndpoint { uri, reply } => {
                    let route_ids = controller.routes_for_endpoint(&uri);
                    if route_ids.is_empty() {
                        let _ =
                            reply.send(Err(CamelError::RouteError("endpoint not found".into())));
                    } else {
                        let health_registry = controller.health_registry();
                        // Detached per-call task: health probes may be slow
                        // and must not block the sequential actor loop.
                        // Unbounded; route via controller JoinSet if QPS grows.
                        tokio::spawn(async move {
                            let futures: Vec<_> = route_ids
                                .iter()
                                .map(|rid| health_registry.check_route(rid))
                                .collect();
                            let results = futures::future::join_all(futures).await;
                            let worst = results.into_iter().fold(
                                camel_api::HealthStatus::Healthy,
                                crate::health_registry::combine_worst,
                            );
                            let _ = reply.send(Ok(worst));
                        });
                    }
                }
                RouteControllerCommand::AutoStartupRouteIds { reply } => {
                    let _ = reply.send(controller.auto_startup_route_ids());
                }
                RouteControllerCommand::ShutdownRouteIds { reply } => {
                    let _ = reply.send(controller.shutdown_route_ids());
                }
                RouteControllerCommand::GetPipeline { route_id, reply } => {
                    let _ = reply.send(controller.get_pipeline(&route_id));
                }
                RouteControllerCommand::StartRouteReload { route_id, reply } => {
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    if restarting
                        .lock()
                        .expect("restarting mutex poisoned") // allow-unwrap
                        .contains(&route_id)
                    {
                        let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                            "route {} is restarting",
                            route_id
                        ))));
                        continue;
                    }
                    let _ = reply.send(controller.start_route_reload(&route_id).await);
                }
                RouteControllerCommand::StopRouteReload { route_id, reply } => {
                    // allow-unwrap: Mutex cannot be poisoned in normal operation
                    if restarting
                        .lock()
                        .expect("restarting mutex poisoned") // allow-unwrap
                        .contains(&route_id)
                    {
                        let _ = reply.send(Err(camel_api::CamelError::RouteError(format!(
                            "route {} is restarting",
                            route_id
                        ))));
                        continue;
                    }
                    let _ = reply.send(controller.stop_route_reload(&route_id).await);
                }
                RouteControllerCommand::SetRuntimeHandle { runtime } => {
                    controller.set_runtime_handle(runtime);
                }
                RouteControllerCommand::SetInterceptRules { rules, reply } => {
                    let _ = reply.send(controller.set_intercept_rules(rules));
                }
                RouteControllerCommand::MarkStarted { reply } => {
                    controller.mark_started();
                    let _ = reply.send(Ok(()));
                }
                RouteControllerCommand::SetFunctionInvoker { invoker } => {
                    controller.set_function_invoker(invoker);
                }
                RouteControllerCommand::RouteSourceHash { route_id, reply } => {
                    let _ = reply.send(controller.route_source_hash(&route_id));
                }
                RouteControllerCommand::RouteHasLifecycle { route_id, reply } => {
                    let _ = reply.send(controller.route_has_lifecycle(&route_id));
                }
                RouteControllerCommand::Shutdown => {
                    break;
                }
            }
        }
    });
    (RouteControllerHandle { tx, cohort }, handle)
}

pub fn spawn_supervision_task(
    controller: RouteControllerHandle,
    config: SupervisionConfig,
    _metrics: Option<Arc<dyn MetricsCollector>>,
    mut crash_rx: mpsc::Receiver<CrashNotification>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut attempts: std::collections::HashMap<String, u32> = std::collections::HashMap::new();
        let mut last_restart_time: std::collections::HashMap<String, Instant> =
            std::collections::HashMap::new();
        let mut currently_restarting: std::collections::HashSet<String> =
            std::collections::HashSet::new();

        debug!("Supervision loop started");

        while let Some(notification) = crash_rx.recv().await {
            let route_id = notification.route_id;
            if currently_restarting.contains(&route_id) {
                continue;
            }

            if let Some(last_time) = last_restart_time.get(&route_id)
                && last_time.elapsed() >= config.initial_delay
            {
                attempts.insert(route_id.clone(), 0);
            }

            let current_attempt = attempts.entry(route_id.clone()).or_insert(0);
            *current_attempt += 1;

            if config
                .max_attempts
                .is_some_and(|max| *current_attempt > max)
            {
                // log-policy: system-broken
                error!(
                    route_id = %route_id,
                    attempts = *current_attempt,
                    "Route exceeded max restart attempts, giving up"
                );
                continue;
            }

            let delay = config.next_delay(*current_attempt);
            currently_restarting.insert(route_id.clone());
            tokio::time::sleep(delay).await;

            match controller.restart_route(route_id.clone()).await {
                Ok(()) => {
                    info!(route_id = %route_id, "Route restarted successfully");
                    last_restart_time.insert(route_id.clone(), Instant::now());
                }
                Err(err) => {
                    // log-policy: system-broken
                    error!(route_id = %route_id, error = %err, "Failed to restart route");
                }
            }

            currently_restarting.remove(&route_id);
        }

        debug!("Supervision loop ended");
    })
}

#[cfg(test)]
mod tests {
    use super::{
        RouteControllerCommand, RouteControllerHandle, spawn_controller_actor,
        spawn_supervision_task,
    };
    use crate::lifecycle::CohortActivationGate;
    use crate::lifecycle::adapters::route_controller::DefaultRouteController;
    use crate::lifecycle::adapters::route_helpers::CrashNotification;
    use crate::lifecycle::application::route_definition::RouteDefinition;
    use crate::shared::components::domain::Registry;
    use crate::shared::observability::domain::TracerConfig;
    use camel_api::function::PrepareToken;
    use camel_api::{
        CamelError, ErrorHandlerConfig, Exchange, ExchangePatch, FunctionDefinition, FunctionDiff,
        FunctionId, FunctionInvocationError, FunctionInvoker, FunctionInvokerSync, RuntimeCommand,
        RuntimeCommandBus, RuntimeCommandResult, RuntimeQuery, RuntimeQueryBus, RuntimeQueryResult,
        SupervisionConfig,
    };
    use std::sync::Arc;
    use std::time::Duration;
    use tokio::sync::mpsc;
    use tokio::time::sleep;
    use tokio::time::timeout;

    fn build_actor_with_components() -> (RouteControllerHandle, tokio::task::JoinHandle<()>) {
        let registry = Arc::new(std::sync::Mutex::new(Registry::new()));
        {
            let mut guard = registry.lock().expect("lock");
            guard.register(std::sync::Arc::new(
                camel_component_timer::TimerComponent::new(),
            ));
            guard.register(std::sync::Arc::new(
                camel_component_mock::MockComponent::new(),
            ));
        }
        let controller = DefaultRouteController::new(
            Arc::clone(&registry),
            Arc::new(camel_api::NoopPlatformService::default()),
        );
        spawn_controller_actor(controller)
    }

    fn build_empty_actor() -> (RouteControllerHandle, tokio::task::JoinHandle<()>) {
        let controller = DefaultRouteController::new(
            Arc::new(std::sync::Mutex::new(Registry::new())),
            Arc::new(camel_api::NoopPlatformService::default()),
        );
        spawn_controller_actor(controller)
    }

    fn route_def(route_id: &str, from_uri: &str) -> RouteDefinition {
        RouteDefinition::new(from_uri, vec![]).with_route_id(route_id)
    }

    struct NoopRuntime;
    struct NoopInvoker;

    #[async_trait::async_trait]
    impl RuntimeCommandBus for NoopRuntime {
        async fn execute(&self, _cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError> {
            Ok(RuntimeCommandResult::Accepted)
        }
    }

    #[async_trait::async_trait]
    impl RuntimeQueryBus for NoopRuntime {
        async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError> {
            Ok(match query {
                RuntimeQuery::GetRouteStatus { route_id }
                | RuntimeQuery::InFlightCount { route_id } => {
                    RuntimeQueryResult::RouteNotFound { route_id }
                }
                // ListRoutes and any future variant return an empty result.
                _ => RuntimeQueryResult::Routes {
                    route_ids: Vec::new(),
                },
            })
        }
    }

    impl FunctionInvokerSync for NoopInvoker {
        fn stage_pending(
            &self,
            _def: FunctionDefinition,
            _route_id: Option<&str>,
            _generation: u64,
        ) {
        }
        fn discard_staging(&self, _generation: u64) {}
        fn begin_reload(&self) -> u64 {
            1
        }
        fn function_refs_for_route(&self, _route_id: &str) -> Vec<(FunctionId, Option<String>)> {
            vec![]
        }
        fn staged_refs_for_route(
            &self,
            _route_id: &str,
            _generation: u64,
        ) -> Vec<(FunctionId, Option<String>)> {
            vec![]
        }
        fn staged_defs_for_route(
            &self,
            _route_id: &str,
            _generation: u64,
        ) -> Vec<(FunctionDefinition, Option<String>)> {
            vec![]
        }
    }

    #[async_trait::async_trait]
    impl FunctionInvoker for NoopInvoker {
        async fn register(
            &self,
            _def: FunctionDefinition,
            _route_id: Option<&str>,
        ) -> Result<(), FunctionInvocationError> {
            Ok(())
        }
        async fn unregister(
            &self,
            _id: &FunctionId,
            _route_id: Option<&str>,
        ) -> Result<(), FunctionInvocationError> {
            Ok(())
        }
        async fn invoke(
            &self,
            _id: &FunctionId,
            _exchange: &Exchange,
        ) -> Result<ExchangePatch, FunctionInvocationError> {
            Ok(ExchangePatch::default())
        }
        async fn prepare_reload(
            &self,
            _diff: FunctionDiff,
            _generation: u64,
        ) -> Result<PrepareToken, FunctionInvocationError> {
            Ok(PrepareToken::default())
        }
        async fn finalize_reload(
            &self,
            _diff: &FunctionDiff,
            _generation: u64,
        ) -> Result<(), FunctionInvocationError> {
            Ok(())
        }
        async fn rollback_reload(
            &self,
            _token: PrepareToken,
            _generation: u64,
        ) -> Result<(), FunctionInvocationError> {
            Ok(())
        }
        async fn commit_staged(&self) -> Result<(), FunctionInvocationError> {
            Ok(())
        }
    }

    #[tokio::test]
    async fn start_route_sends_command_and_returns_reply() {
        let (tx, mut rx) = mpsc::channel(1);
        let handle = RouteControllerHandle {
            tx,
            cohort: Arc::new(CohortActivationGate::new_closed()),
        };

        let task = tokio::spawn(async move { handle.start_route("route-a").await });

        let command = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("command should be received within 2s")
            .expect("command channel alive");
        match command {
            RouteControllerCommand::StartRoute { route_id, reply } => {
                assert_eq!(route_id, "route-a");
                let _ = reply.send(Ok(()));
            }
            _ => panic!("unexpected command variant"),
        }

        let result = task.await.expect("join should succeed");
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn start_route_returns_error_when_actor_stops() {
        let (tx, rx) = mpsc::channel(1);
        drop(rx);

        let handle = RouteControllerHandle {
            tx,
            cohort: Arc::new(CohortActivationGate::new_closed()),
        };
        let result = handle.start_route("route-a").await;

        assert!(matches!(result, Err(CamelError::ProcessorError(_))));
    }

    #[tokio::test]
    async fn spawn_controller_actor_processes_commands_and_shutdown() {
        let controller = DefaultRouteController::new(
            Arc::new(std::sync::Mutex::new(Registry::new())),
            Arc::new(camel_api::NoopPlatformService::default()),
        );
        let (handle, join_handle) = spawn_controller_actor(controller);

        assert_eq!(handle.route_count().await.expect("route_count"), 0);
        assert_eq!(
            handle.route_ids().await.expect("route_ids"),
            Vec::<String>::new()
        );

        handle.shutdown().await.expect("shutdown send");
        join_handle.await.expect("actor join");
    }

    #[tokio::test]
    async fn actor_handle_introspection_and_mutation_commands() {
        let (handle, join_handle) = build_actor_with_components();
        let definition = route_def("h-1", "timer:tick?period=100");

        handle.add_route(definition).await.expect("add route");
        assert!(handle.route_exists("h-1").await.expect("route exists h-1"));
        assert!(
            !handle
                .route_exists("no-such")
                .await
                .expect("route exists no-such")
        );

        let from_uri = handle.route_from_uri("h-1").await.expect("route_from_uri");
        assert_eq!(from_uri.as_deref(), Some("timer:tick?period=100"));
        assert_eq!(handle.route_count().await.expect("route_count"), 1);

        let auto_ids = handle
            .auto_startup_route_ids()
            .await
            .expect("auto_startup_route_ids");
        assert!(auto_ids.iter().any(|id| id == "h-1"));

        let shutdown_ids = handle
            .shutdown_route_ids()
            .await
            .expect("shutdown_route_ids");
        assert!(shutdown_ids.iter().any(|id| id == "h-1"));

        let compiled = handle
            .compile_route_definition(route_def("h-1", "timer:tick?period=100"))
            .await
            .expect("compile_route_definition");

        assert!(
            handle
                .get_pipeline("h-1")
                .await
                .expect("get_pipeline")
                .is_some()
        );
        handle
            .swap_pipeline("h-1", compiled)
            .await
            .expect("swap_pipeline");

        let _ = handle
            .in_flight_count("h-1")
            .await
            .expect("in_flight_count");
        let _ = handle.route_source_hash("h-1").await;

        handle
            .set_error_handler(ErrorHandlerConfig::dead_letter_channel("log:dlq"))
            .await
            .expect("set_error_handler");
        handle
            .set_tracer_config(TracerConfig::default())
            .await
            .expect("set_tracer_config");
        handle
            .set_runtime_handle(Arc::new(NoopRuntime))
            .await
            .expect("set_runtime_handle");

        handle.remove_route("h-1").await.expect("remove_route");
        assert_eq!(
            handle
                .route_count()
                .await
                .expect("route_count after remove"),
            0
        );
        handle
            .stop_all_routes()
            .await
            .expect("stop_all_routes on empty");

        handle.shutdown().await.expect("shutdown send");
        join_handle.await.expect("actor join");
    }

    #[tokio::test]
    async fn actor_handle_lifecycle_start_stop_restart_suspend_resume() {
        let (handle, join_handle) = build_actor_with_components();
        handle
            .add_route(route_def("lc-1", "timer:tick?period=50"))
            .await
            .expect("add route lc-1");

        handle.start_route("lc-1").await.expect("start_route");
        sleep(Duration::from_millis(20)).await;

        handle.restart_route("lc-1").await.expect("restart_route");
        sleep(Duration::from_millis(20)).await;

        handle.suspend_route("lc-1").await.expect("suspend_route");
        handle.resume_route("lc-1").await.expect("resume_route");
        sleep(Duration::from_millis(20)).await;

        handle.stop_route("lc-1").await.expect("stop_route");
        handle.start_all_routes().await.expect("start_all_routes");
        sleep(Duration::from_millis(20)).await;
        handle.stop_all_routes().await.expect("stop_all_routes");

        handle
            .start_route_reload("lc-1")
            .await
            .expect("start_route_reload");
        handle
            .stop_route_reload("lc-1")
            .await
            .expect("stop_route_reload");

        handle.shutdown().await.expect("shutdown send");
        join_handle.await.expect("actor join");
    }

    #[tokio::test]
    async fn spawn_supervision_restarts_route_on_crash() {
        let (handle, join_handle) = build_actor_with_components();
        handle
            .add_route(route_def("sup-1", "timer:tick?period=100"))
            .await
            .expect("add route sup-1");
        handle
            .start_route("sup-1")
            .await
            .expect("start_route sup-1");

        let (crash_tx, crash_rx) = mpsc::channel(8);
        let supervision = spawn_supervision_task(
            handle.clone(),
            SupervisionConfig {
                initial_delay: Duration::from_millis(10),
                max_attempts: Some(2),
                ..SupervisionConfig::default()
            },
            None,
            crash_rx,
        );

        crash_tx
            .send(CrashNotification {
                route_id: "sup-1".to_string(),
                error: "simulated".to_string(),
            })
            .await
            .expect("send crash notification");

        sleep(Duration::from_millis(150)).await;
        drop(crash_tx);
        supervision.await.expect("supervision join");

        handle.shutdown().await.expect("shutdown send");
        join_handle.await.expect("actor join");
    }

    #[tokio::test]
    async fn supervision_skips_duplicate_and_gives_up_after_max_attempts() {
        let (handle, join_handle) = build_actor_with_components();
        handle
            .add_route(route_def("sup-2", "timer:tick?period=100"))
            .await
            .expect("add route sup-2");
        handle
            .start_route("sup-2")
            .await
            .expect("start_route sup-2");

        let (crash_tx, crash_rx) = mpsc::channel(8);
        let supervision = spawn_supervision_task(
            handle.clone(),
            SupervisionConfig {
                initial_delay: Duration::from_millis(10),
                max_attempts: Some(1),
                ..SupervisionConfig::default()
            },
            None,
            crash_rx,
        );

        crash_tx
            .send(CrashNotification {
                route_id: "sup-2".to_string(),
                error: "attempt-1".to_string(),
            })
            .await
            .expect("send crash attempt-1");
        crash_tx
            .send(CrashNotification {
                route_id: "sup-2".to_string(),
                error: "attempt-2".to_string(),
            })
            .await
            .expect("send crash attempt-2");

        sleep(Duration::from_millis(200)).await;
        drop(crash_tx);
        supervision.await.expect("supervision join");

        handle.shutdown().await.expect("shutdown send");
        join_handle.await.expect("actor join");
    }

    #[tokio::test]
    async fn try_set_runtime_handle_succeeds_on_fresh_actor() {
        let (handle, join_handle) = build_empty_actor();

        handle
            .try_set_runtime_handle(Arc::new(NoopRuntime))
            .expect("try_set_runtime_handle should succeed");

        handle.shutdown().await.expect("shutdown send");
        join_handle.await.expect("actor join");
    }

    #[tokio::test]
    async fn shutdown_returns_error_when_actor_stopped() {
        let (tx, rx) = mpsc::channel(1);
        drop(rx);

        let handle = RouteControllerHandle {
            tx,
            cohort: Arc::new(CohortActivationGate::new_closed()),
        };
        let result = handle.shutdown().await;

        assert!(matches!(result, Err(CamelError::ProcessorError(_))));
    }

    #[tokio::test]
    async fn handle_methods_send_expected_commands_and_receive_replies() {
        let (tx, mut rx) = mpsc::channel(16);
        let handle = RouteControllerHandle {
            tx,
            cohort: Arc::new(CohortActivationGate::new_closed()),
        };

        let stop_task = tokio::spawn({
            let h = handle.clone();
            async move { h.stop_route("r-1").await }
        });
        let cmd = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("stop command within 2s")
            .expect("command channel alive");
        match cmd {
            RouteControllerCommand::StopRoute { route_id, reply } => {
                assert_eq!(route_id, "r-1");
                let _ = reply.send(Ok(()));
            }
            _ => panic!("unexpected command"),
        }
        assert!(stop_task.await.expect("join").is_ok());

        let exists_task = tokio::spawn({
            let h = handle.clone();
            async move { h.route_exists("r-2").await }
        });
        let cmd = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("exists command within 2s")
            .expect("command channel alive");
        match cmd {
            RouteControllerCommand::RouteExists { route_id, reply } => {
                assert_eq!(route_id, "r-2");
                let _ = reply.send(true);
            }
            _ => panic!("unexpected command"),
        }
        assert!(exists_task.await.expect("join").expect("ok"));

        let hash_task = tokio::spawn({
            let h = handle.clone();
            async move { h.route_source_hash("r-3").await }
        });
        let cmd = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("hash command within 2s")
            .expect("command channel alive");
        match cmd {
            RouteControllerCommand::RouteSourceHash { route_id, reply } => {
                assert_eq!(route_id, "r-3");
                let _ = reply.send(Some(77));
            }
            _ => panic!("unexpected command"),
        }
        assert_eq!(hash_task.await.expect("join"), Some(77));
    }

    #[tokio::test]
    async fn handle_methods_error_on_dropped_reply_channel() {
        let (tx, mut rx) = mpsc::channel(16);
        let handle = RouteControllerHandle {
            tx,
            cohort: Arc::new(CohortActivationGate::new_closed()),
        };

        let count_task = tokio::spawn({
            let h = handle.clone();
            async move { h.route_count().await }
        });
        let cmd = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("route_count command within 2s")
            .expect("command channel alive");
        match cmd {
            RouteControllerCommand::RouteCount { reply } => drop(reply),
            _ => panic!("unexpected command"),
        }
        assert!(matches!(
            count_task.await.expect("join"),
            Err(CamelError::ProcessorError(_))
        ));

        let stop_task = tokio::spawn({
            let h = handle.clone();
            async move { h.stop_route("x").await }
        });
        let cmd = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("stop command within 2s")
            .expect("command channel alive");
        match cmd {
            RouteControllerCommand::StopRoute { reply, .. } => drop(reply),
            _ => panic!("unexpected command"),
        }
        assert!(matches!(
            stop_task.await.expect("join"),
            Err(CamelError::ProcessorError(_))
        ));

        let maybe_hash = tokio::spawn({
            let h = handle.clone();
            async move { h.route_source_hash("x").await }
        });
        let cmd = timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("hash command within 2s")
            .expect("command channel alive");
        match cmd {
            RouteControllerCommand::RouteSourceHash { reply, .. } => drop(reply),
            _ => panic!("unexpected command"),
        }
        assert_eq!(maybe_hash.await.expect("join"), None);
    }

    #[test]
    fn try_set_function_invoker_returns_mailbox_full() {
        let (tx, mut rx) = mpsc::channel(1);
        tx.try_send(RouteControllerCommand::Shutdown)
            .expect("fill mailbox");
        let handle = RouteControllerHandle {
            tx,
            cohort: Arc::new(CohortActivationGate::new_closed()),
        };

        let result = handle.try_set_function_invoker(Arc::new(NoopInvoker));
        assert!(matches!(result, Err(CamelError::ProcessorError(_))));

        rx.try_recv().expect("mailbox still has first message");
    }

    #[tokio::test]
    async fn test_restart_does_not_block_other_routes() {
        // D-L6: a route's restart (stop + 100ms sleep + start) must not block
        // commands for other routes. Without the fix, start_c would wait
        // behind restart's 100ms sleep in the actor mailbox.
        let (handle, join_handle) = build_actor_with_components();
        handle
            .add_route(route_def("route-a", "timer:tick?period=100"))
            .await
            .expect("add route-a");
        handle
            .add_route(route_def("route-b", "timer:tick?period=100"))
            .await
            .expect("add route-b");
        handle
            .add_route(route_def("route-c", "timer:tick?period=100"))
            .await
            .expect("add route-c");

        handle.start_route("route-a").await.expect("start route-a");
        handle.start_route("route-b").await.expect("start route-b");

        // Spawn restart on its own task so its RestartRoute command is
        // guaranteed to land in the mailbox before start_c is enqueued.
        let restart_handle = handle.clone();
        let restart_fut =
            tokio::spawn(async move { restart_handle.restart_route("route-a").await });

        // Yield once so restart's send completes and RestartRoute is queued first.
        tokio::task::yield_now().await;

        // Now issue start_c — its StartRoute command is queued behind RestartRoute.
        let start_c_t0 = std::time::Instant::now();
        let start_result = handle.start_route("route-c").await;
        let start_c_elapsed = start_c_t0.elapsed();

        assert!(start_result.is_ok(), "route-c start should succeed");
        assert!(
            start_c_elapsed < Duration::from_millis(80),
            "route-c start blocked by route-a restart (took {start_c_elapsed:?}); \
             expected to complete before restart's 100ms sleep"
        );

        let restart_result = restart_fut.await.expect("restart join");
        assert!(restart_result.is_ok(), "route-a restart should succeed");

        handle.shutdown().await.expect("shutdown send");
        join_handle.await.expect("actor join");
    }

    #[tokio::test]
    async fn test_command_to_restarting_route_is_rejected() {
        // D-L6: while a route is restarting, mutating commands for that route
        // must be rejected.
        let (handle, join_handle) = build_actor_with_components();
        handle
            .add_route(route_def("route-a", "timer:tick?period=100"))
            .await
            .expect("add route-a");
        handle.start_route("route-a").await.expect("start route-a");

        // Spawn restart on its own task so the RestartRoute command is
        // guaranteed to land in the mailbox.
        let restart_handle = handle.clone();
        let restart_fut =
            tokio::spawn(async move { restart_handle.restart_route("route-a").await });

        // Give the actor time to process the Restart and enter the 100ms sleep.
        // After the inline stop completes, the actor returns to rx.recv() and
        // the spawned task is sleeping — route-a is in the restarting set.
        tokio::time::sleep(Duration::from_millis(20)).await;

        // While restarting, try to stop route-a — must be rejected.
        let stop_result = handle.stop_route("route-a").await;
        assert!(
            stop_result.is_err(),
            "stop during restart should be rejected"
        );
        assert!(
            stop_result.unwrap_err().to_string().contains("restarting"),
            "error should mention restarting"
        );

        // Now the restart completes.
        let restart_result = restart_fut.await.expect("restart join");
        assert!(restart_result.is_ok(), "route-a restart should succeed");

        handle.shutdown().await.expect("shutdown send");
        join_handle.await.expect("actor join");
    }
}