rust_supervisor/

lib.rs

//! # rust_supervisor
//! 
//! `rust_supervisor` is a library inspired by Erlang/OTP's supervision system,
//! allowing automatic process restart when they fail.
//!
//! ## Main features
//!
//! * Multiple restart strategies (OneForOne, OneForAll, RestForOne)
//! * Flexible restart policy configuration
//! * Process dependency management
//! * Automatic process state monitoring
//! * Hierarchical supervision (supervisors supervising supervisors)
//! * Child specs with permanent/temporary/transient types
//! * Graceful shutdown with timeout
//! * Event callbacks for observability

use std::collections::HashMap;
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use std::time::{Duration, Instant};
use std::sync::atomic::{AtomicBool, Ordering};

/// Defines the strategy to use when a process fails
#[derive(Debug, Clone)]
pub enum RestartStrategy {
    /// Restart only the failed process
    OneForOne,
    /// Restart all processes when one fails
    OneForAll,
    /// Restart the failed process and all processes that depend on it
    RestForOne,
}

/// Represents the current state of a process
#[derive(Debug, Clone, PartialEq)]
pub enum ProcessState {
    /// Process is running
    Running,
    /// Process has failed
    Failed,
    /// Process is being restarted
    Restarting,
    /// Process is stopped (will not be restarted)
    Stopped,
    /// Process is not yet started
    Unstarted,
}

/// Child specification defining how a process should be supervised
#[derive(Debug, Clone)]
pub enum ChildType {
    /// Process is always restarted on failure
    Permanent,
    /// Process is never restarted on failure
    Temporary,
    /// Process is restarted only if it didn't exit normally
    Transient,
}

/// Shutdown strategy for gracefully stopping processes
#[derive(Debug, Clone)]
pub enum ShutdownStrategy {
    /// Kill the process immediately
    BrutalKill,
    /// Wait for graceful shutdown with timeout
    Shutdown(Duration),
}

/// Supervisor configuration
#[derive(Debug, Clone)]
pub struct SupervisorConfig {
    /// Maximum number of restarts allowed
    pub max_restarts: usize,
    /// Time period over which to count restarts
    pub max_time: Duration,
    /// Restart strategy to use
    pub restart_strategy: RestartStrategy,
    /// Default shutdown strategy for children
    pub shutdown_strategy: ShutdownStrategy,
}

impl Default for SupervisorConfig {
    /// Creates a default configuration with reasonable values
    fn default() -> Self {
        SupervisorConfig {
            max_restarts: 3,
            max_time: Duration::from_secs(5),
            restart_strategy: RestartStrategy::OneForOne,
            shutdown_strategy: ShutdownStrategy::Shutdown(Duration::from_secs(5)),
        }
    }
}

/// Event callback for observability
pub trait EventCallback: Send + Sync {
    /// Called when a process starts
    fn on_process_started(&self, _process_name: &str) {}
    /// Called when a process fails
    fn on_process_failed(&self, _process_name: &str) {}
    /// Called when a process is restarted
    fn on_process_restarted(&self, _process_name: &str, _restart_count: usize) {}
    /// Called when a process is stopped
    fn on_process_stopped(&self, _process_name: &str) {}
}

/// Default no-op event callback
pub struct NoOpCallback;

impl EventCallback for NoOpCallback {}

/// Child process specification
struct ChildSpec {
    /// Type of child (Permanent, Temporary, Transient)
    child_type: ChildType,
    /// Factory for creating a new instance of the process
    factory: Box<dyn Fn() -> thread::JoinHandle<()> + Send + 'static>,
    /// Shutdown strategy for this child
    shutdown_strategy: ShutdownStrategy,
    /// Flag to signal graceful shutdown
    shutdown_signal: Arc<AtomicBool>,
}

/// Internal information about a managed process
struct ProcessInfo {
    /// Handle to the running thread (None if not started or failed)
    handle: Option<thread::JoinHandle<()>>,
    /// Restart history for applying the limiting policy
    restart_times: Vec<Instant>,
    /// Current process state
    state: ProcessState,
    /// Number of restarts since startup
    restart_count: usize,
    /// Child specification
    spec: ChildSpec,
}

/// Supervisor that manages a set of processes
pub struct Supervisor {
    /// Map of managed processes, with their name as the key
    processes: Arc<Mutex<HashMap<String, ProcessInfo>>>,
    /// Supervisor configuration
    config: SupervisorConfig,
    /// Map of dependencies between processes
    dependencies: Arc<Mutex<HashMap<String, Vec<String>>>>,
    /// Event callback for observability
    event_callback: Arc<dyn EventCallback>,
    /// Monitoring thread handle
    monitor_handle: Arc<Mutex<Option<thread::JoinHandle<()>>>>,
    /// Flag to signal monitoring thread to stop
    shutdown_flag: Arc<AtomicBool>,
    /// Channel for internal signaling
    signal_tx: Arc<Mutex<Option<mpsc::Sender<()>>>>,
}

impl Supervisor {
    /// Creates a new supervisor with the specified configuration
    ///
    /// # Arguments
    ///
    /// * `config` - Supervisor configuration
    ///
    /// # Example
    ///
    /// ```ignore
    /// use rust_supervisor::{Supervisor, SupervisorConfig};
    /// let supervisor = Supervisor::new(SupervisorConfig::default());
    /// ```
    pub fn new(config: SupervisorConfig) -> Self {
        Supervisor::with_callback(config, Arc::new(NoOpCallback))
    }

    /// Creates a new supervisor with a custom event callback
    pub fn with_callback(config: SupervisorConfig, callback: Arc<dyn EventCallback>) -> Self {
        Supervisor {
            processes: Arc::new(Mutex::new(HashMap::new())),
            config,
            dependencies: Arc::new(Mutex::new(HashMap::new())),
            event_callback: callback,
            monitor_handle: Arc::new(Mutex::new(None)),
            shutdown_flag: Arc::new(AtomicBool::new(false)),
            signal_tx: Arc::new(Mutex::new(None)),
        }
    }

    /// Adds a process to monitor
    ///
    /// # Arguments
    ///
    /// * `name` - Unique process name
    /// * `child_type` - Type of child (Permanent, Temporary, Transient)
    /// * `factory` - Function that creates and starts the process
    ///
    /// # Example
    ///
    /// ```ignore
    /// use rust_supervisor::{Supervisor, SupervisorConfig, ChildType};
    /// use std::thread;
    /// 
    /// let mut supervisor = Supervisor::new(SupervisorConfig::default());
    /// supervisor.add_process("worker", ChildType::Permanent, || {
    ///     thread::spawn(|| {
    ///         // Worker code...
    ///     })
    /// });
    /// ```
    pub fn add_process<F>(&mut self, name: &str, child_type: ChildType, factory: F)
    where
        F: Fn() -> thread::JoinHandle<()> + Send + 'static,
    {
        self.add_process_with_shutdown(
            name,
            child_type,
            factory,
            self.config.shutdown_strategy.clone(),
        );
    }

    /// Adds a process with custom shutdown strategy
    pub fn add_process_with_shutdown<F>(
        &mut self,
        name: &str,
        child_type: ChildType,
        factory: F,
        shutdown_strategy: ShutdownStrategy,
    )
    where
        F: Fn() -> thread::JoinHandle<()> + Send + 'static,
    {
        let factory_box = Box::new(factory);
        let shutdown_signal = Arc::new(AtomicBool::new(false));

        let spec = ChildSpec {
            child_type,
            factory: factory_box,
            shutdown_strategy,
            shutdown_signal,
        };

        let mut processes = self.processes.lock().unwrap();
        processes.insert(
            name.to_string(),
            ProcessInfo {
                handle: None,
                restart_times: Vec::new(),
                state: ProcessState::Unstarted,
                restart_count: 0,
                spec,
            },
        );
    }

    /// Declares a dependency between two processes
    ///
    /// # Arguments
    ///
    /// * `process` - Name of the process that depends on another
    /// * `depends_on` - Name of the process that the first one depends on
    pub fn add_dependency(&self, process: &str, depends_on: &str) {
        let mut dependencies = self.dependencies.lock().unwrap();
        dependencies
            .entry(process.to_string())
            .or_insert_with(Vec::new)
            .push(depends_on.to_string());
    }

    /// Starts monitoring processes
    ///
    /// This method launches a monitoring thread that periodically checks
    /// the state of processes and restarts them according to the configured strategy.
    /// 
    /// Returns Arc<Self> for convenience chaining.
    pub fn start_monitoring(self) -> Arc<Self> 
    where
        Self: Sized,
    {
        let supervisor = Arc::new(self);
        
        // Prevent double-start
        let should_start = {
            let handle = supervisor.monitor_handle.lock().unwrap();
            handle.is_none()
        };

        if !should_start {
            return supervisor;
        }

        // Setup monitoring thread
        {
            let (tx, _rx) = mpsc::channel();
            *supervisor.signal_tx.lock().unwrap() = Some(tx);

            let supervisor_clone = Arc::clone(&supervisor);
            let monitor_thread = thread::spawn(move || {
                supervisor_clone.monitor_loop();
            });

            let mut handle = supervisor.monitor_handle.lock().unwrap();
            *handle = Some(monitor_thread);
        }

        // Start all processes initially
        {
            let mut processes = supervisor.processes.lock().unwrap();
            for (name, info) in processes.iter_mut() {
                info.state = ProcessState::Restarting;
                info.handle = Some((info.spec.factory)());
                info.state = ProcessState::Running;
                info.restart_times.push(Instant::now());
                supervisor.event_callback.on_process_started(name);
            }
        }

        supervisor
    }

    /// Internal monitoring loop
    fn monitor_loop(&self) {
        loop {
            if self.shutdown_flag.load(Ordering::Relaxed) {
                break;
            }

            thread::sleep(Duration::from_millis(100));

            // Collect failed processes
            let mut failed_processes = Vec::new();
            {
                let mut processes = self.processes.lock().unwrap();
                for (name, info) in processes.iter_mut() {
                    if info.state == ProcessState::Unstarted {
                        continue;
                    }

                    if let Some(handle) = &info.handle {
                        if handle.is_finished() {
                            info.state = ProcessState::Failed;
                            info.handle = None;
                            self.event_callback.on_process_failed(name);

                            // Check if we should restart based on child type
                            let should_check_restart = match info.spec.child_type {
                                ChildType::Permanent => true,
                                ChildType::Temporary => false,
                                ChildType::Transient => {
                                    // In transient mode, restart only if exit was abnormal
                                    // (In this simplified version, we assume abnormal exit)
                                    true
                                }
                            };

                            if should_check_restart {
                                let now = Instant::now();
                                info.restart_times
                                    .retain(|time| now.duration_since(*time) < self.config.max_time);

                                if info.restart_times.len() < self.config.max_restarts {
                                    failed_processes.push(name.clone());
                                } else {
                                    info.state = ProcessState::Stopped;
                                }
                            } else {
                                info.state = ProcessState::Stopped;
                            }
                        }
                    }
                }
            }

            // Handle restarts based on strategy
            for failed_process in failed_processes {
                let processes_to_restart = {
                    let processes = self.processes.lock().unwrap();
                    let dependencies = self.dependencies.lock().unwrap();

                    match self.config.restart_strategy {
                        RestartStrategy::OneForOne => vec![failed_process.clone()],
                        RestartStrategy::OneForAll => processes.keys().cloned().collect(),
                        RestartStrategy::RestForOne => {
                            let mut to_restart = vec![failed_process.clone()];
                            for (proc_name, deps) in dependencies.iter() {
                                if deps.contains(&failed_process) {
                                    to_restart.push(proc_name.clone());
                                }
                            }
                            to_restart
                        }
                    }
                };

                let now = Instant::now();
                for proc_name in processes_to_restart {
                    let mut processes = self.processes.lock().unwrap();
                    if let Some(proc_info) = processes.get_mut(&proc_name) {
                        // Skip if Temporary or already Stopped
                        if matches!(proc_info.spec.child_type, ChildType::Temporary)
                            || proc_info.state == ProcessState::Stopped
                        {
                            continue;
                        }

                        proc_info.state = ProcessState::Restarting;
                        proc_info.restart_count += 1;
                        proc_info.handle = Some((proc_info.spec.factory)());
                        proc_info.restart_times.push(now);
                        proc_info.state = ProcessState::Running;

                        self.event_callback
                            .on_process_restarted(&proc_name, proc_info.restart_count);
                    }
                }
            }
        }
    }

    /// Manually stops a process with graceful shutdown
    ///
    /// # Arguments
    ///
    /// * `name` - Name of the process to stop
    ///
    /// # Returns
    ///
    /// `true` if the process was found and stopped, `false` otherwise
    pub fn stop_process(&self, name: &str) -> bool {
        let mut processes = self.processes.lock().unwrap();
        if let Some(info) = processes.get_mut(name) {
            if let Some(handle) = info.handle.take() {
                // Signal graceful shutdown
                info.spec.shutdown_signal.store(true, Ordering::Relaxed);

                match &info.spec.shutdown_strategy {
                    ShutdownStrategy::BrutalKill => {
                        drop(handle);
                    }
                    ShutdownStrategy::Shutdown(timeout) => {
                        // Try to wait for graceful shutdown
                        let start = Instant::now();
                        while !handle.is_finished() && start.elapsed() < *timeout {
                            thread::sleep(Duration::from_millis(10));
                        }
                        drop(handle);
                    }
                }

                info.state = ProcessState::Stopped;
                self.event_callback.on_process_stopped(name);
                return true;
            }
        }
        false
    }

    /// Stops all processes gracefully
    pub fn shutdown(&self) {
        self.shutdown_flag.store(true, Ordering::Relaxed);

        let process_names: Vec<String> = {
            let processes = self.processes.lock().unwrap();
            processes.keys().cloned().collect()
        };

        for name in process_names {
            self.stop_process(&name);
        }

        // Wait for monitoring thread to finish
        if let Ok(mut handle) = self.monitor_handle.lock() {
            if let Some(thread) = handle.take() {
                let _ = thread.join();
            }
        }
    }

    /// Gets the current state of a process
    ///
    /// # Arguments
    ///
    /// * `name` - Process name
    ///
    /// # Returns
    ///
    /// The process state, or `None` if the process doesn't exist
    pub fn get_process_state(&self, name: &str) -> Option<ProcessState> {
        let processes = self.processes.lock().unwrap();
        processes.get(name).map(|info| info.state.clone())
    }

    /// Gets the restart count for a process
    pub fn get_restart_count(&self, name: &str) -> Option<usize> {
        let processes = self.processes.lock().unwrap();
        processes.get(name).map(|info| info.restart_count)
    }

    /// Gets all process states
    pub fn get_all_states(&self) -> HashMap<String, (ProcessState, usize)> {
        let processes = self.processes.lock().unwrap();
        processes
            .iter()
            .map(|(name, info)| {
                (
                    name.clone(),
                    (info.state.clone(), info.restart_count),
                )
            })
            .collect()
    }

    /// Get the shutdown signal for a process (for graceful shutdown detection)
    pub fn get_shutdown_signal(&self, name: &str) -> Option<Arc<AtomicBool>> {
        let processes = self.processes.lock().unwrap();
        processes.get(name).map(|info| Arc::clone(&info.spec.shutdown_signal))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// Test that a supervisor can be created with default config
    #[test]
    fn test_supervisor_creation() {
        let supervisor = Supervisor::new(SupervisorConfig::default());
        assert_eq!(supervisor.get_all_states().len(), 0);
    }

    /// Test adding a process to supervisor
    #[test]
    fn test_add_process() {
        let mut supervisor = Supervisor::new(SupervisorConfig::default());
        supervisor.add_process("worker1", ChildType::Permanent, || {
            thread::spawn(|| {
                thread::sleep(Duration::from_secs(10));
            })
        });

        assert_eq!(supervisor.get_all_states().len(), 1);
        assert_eq!(
            supervisor.get_process_state("worker1"),
            Some(ProcessState::Unstarted)
        );
    }

    /// Test that processes start when monitoring is started
    #[test]
    fn test_process_starts_on_monitoring() {
        let mut supervisor = Supervisor::new(SupervisorConfig::default());
        supervisor.add_process("worker1", ChildType::Permanent, || {
            thread::spawn(|| {
                thread::sleep(Duration::from_secs(10));
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(200));

        assert_eq!(
            supervisor.get_process_state("worker1"),
            Some(ProcessState::Running)
        );

        supervisor.shutdown();
    }

    /// Test permanent process restart on failure
    #[test]
    fn test_permanent_process_restart() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = Arc::clone(&counter);

        let mut supervisor = Supervisor::new(SupervisorConfig::default());
        supervisor.add_process("failing_worker", ChildType::Permanent, move || {
            let cnt = Arc::clone(&counter_clone);
            thread::spawn(move || {
                cnt.fetch_add(1, Ordering::Relaxed);
                panic!("Intentional failure");
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(500));

        // Should have been started and restarted at least once
        assert!(counter.load(Ordering::Relaxed) > 1);

        supervisor.shutdown();
    }

    /// Test temporary process is not restarted
    #[test]
    fn test_temporary_process_no_restart() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = Arc::clone(&counter);

        let mut supervisor = Supervisor::new(SupervisorConfig::default());
        supervisor.add_process("temp_worker", ChildType::Temporary, move || {
            let cnt = Arc::clone(&counter_clone);
            thread::spawn(move || {
                cnt.fetch_add(1, Ordering::Relaxed);
                panic!("Intentional failure");
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(500));

        // Should be started only once, never restarted
        assert_eq!(counter.load(Ordering::Relaxed), 1);
        assert_eq!(
            supervisor.get_process_state("temp_worker"),
            Some(ProcessState::Stopped)
        );

        supervisor.shutdown();
    }

    /// Test stopping a process
    #[test]
    fn test_stop_process() {
        let mut supervisor = Supervisor::new(SupervisorConfig::default());
        supervisor.add_process("worker1", ChildType::Permanent, || {
            thread::spawn(|| {
                thread::sleep(Duration::from_secs(10));
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(200));

        assert!(supervisor.stop_process("worker1"));
        thread::sleep(Duration::from_millis(100));
        assert_eq!(
            supervisor.get_process_state("worker1"),
            Some(ProcessState::Stopped)
        );

        supervisor.shutdown();
    }

    /// Test restart count tracking
    #[test]
    fn test_restart_count() {
        let mut supervisor = Supervisor::new(SupervisorConfig::default());
        supervisor.add_process("failing_worker", ChildType::Permanent, || {
            thread::spawn(|| {
                panic!("Intentional failure");
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(500));

        let restart_count = supervisor.get_restart_count("failing_worker").unwrap_or(0);
        assert!(restart_count > 0);

        supervisor.shutdown();
    }

    /// Test OneForOne restart strategy
    #[test]
    fn test_restart_strategy_one_for_one() {
        let mut config = SupervisorConfig::default();
        config.restart_strategy = RestartStrategy::OneForOne;

        let counter1 = Arc::new(AtomicUsize::new(0));
        let counter1_clone = Arc::clone(&counter1);
        let counter2 = Arc::new(AtomicUsize::new(0));
        let counter2_clone = Arc::clone(&counter2);

        let mut supervisor = Supervisor::new(config);

        supervisor.add_process("failing_worker", ChildType::Permanent, move || {
            let cnt = Arc::clone(&counter1_clone);
            thread::spawn(move || {
                cnt.fetch_add(1, Ordering::Relaxed);
                panic!("Intentional failure");
            })
        });

        supervisor.add_process("stable_worker", ChildType::Permanent, move || {
            let cnt = Arc::clone(&counter2_clone);
            thread::spawn(move || {
                cnt.fetch_add(1, Ordering::Relaxed);
                thread::sleep(Duration::from_secs(10));
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(500));

        let count1 = counter1.load(Ordering::Relaxed);
        let count2 = counter2.load(Ordering::Relaxed);

        // With OneForOne, only failing_worker should be restarted multiple times
        assert!(count1 > count2);

        supervisor.shutdown();
    }

    /// Test process dependencies
    #[test]
    fn test_process_dependencies() {
        let mut supervisor = Supervisor::new(SupervisorConfig::default());

        supervisor.add_process("base_worker", ChildType::Permanent, || {
            thread::spawn(|| {
                thread::sleep(Duration::from_secs(10));
            })
        });

        supervisor.add_process("dependent_worker", ChildType::Permanent, || {
            thread::spawn(|| {
                thread::sleep(Duration::from_secs(10));
            })
        });

        supervisor.add_dependency("dependent_worker", "base_worker");

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(200));

        assert_eq!(
            supervisor.get_process_state("base_worker"),
            Some(ProcessState::Running)
        );
        assert_eq!(
            supervisor.get_process_state("dependent_worker"),
            Some(ProcessState::Running)
        );

        supervisor.shutdown();
    }

    /// Test max restarts limit
    #[test]
    fn test_max_restarts_limit() {
        let mut config = SupervisorConfig::default();
        config.max_restarts = 2;
        config.max_time = Duration::from_secs(5);
        let max_restarts = config.max_restarts;

        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = Arc::clone(&counter);

        let mut supervisor = Supervisor::new(config);
        supervisor.add_process("failing_worker", ChildType::Permanent, move || {
            let cnt = Arc::clone(&counter_clone);
            thread::spawn(move || {
                cnt.fetch_add(1, Ordering::Relaxed);
                panic!("Intentional failure");
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(1000));

        // After reaching max_restarts, process should be stopped
        assert_eq!(
            supervisor.get_process_state("failing_worker"),
            Some(ProcessState::Stopped)
        );

        // Should have been started once + max_restarts attempts
        assert!(counter.load(Ordering::Relaxed) <= max_restarts + 1);

        supervisor.shutdown();
    }

    /// Test graceful shutdown of supervisor
    #[test]
    fn test_supervisor_shutdown() {
        let mut supervisor = Supervisor::new(SupervisorConfig::default());

        supervisor.add_process("worker1", ChildType::Permanent, || {
            thread::spawn(|| {
                thread::sleep(Duration::from_secs(10));
            })
        });

        supervisor.add_process("worker2", ChildType::Permanent, || {
            thread::spawn(|| {
                thread::sleep(Duration::from_secs(10));
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(200));

        supervisor.shutdown();
        thread::sleep(Duration::from_millis(200));

        assert_eq!(
            supervisor.get_process_state("worker1"),
            Some(ProcessState::Stopped)
        );
        assert_eq!(
            supervisor.get_process_state("worker2"),
            Some(ProcessState::Stopped)
        );
    }

    /// Test event callback
    #[test]
    fn test_event_callback() {
        struct TestCallback {
            started: AtomicUsize,
            failed: AtomicUsize,
            restarted: AtomicUsize,
        }

        impl EventCallback for TestCallback {
            fn on_process_started(&self, _process_name: &str) {
                self.started.fetch_add(1, Ordering::Relaxed);
            }

            fn on_process_failed(&self, _process_name: &str) {
                self.failed.fetch_add(1, Ordering::Relaxed);
            }

            fn on_process_restarted(&self, _process_name: &str, _restart_count: usize) {
                self.restarted.fetch_add(1, Ordering::Relaxed);
            }
        }

        let callback: Arc<dyn EventCallback> = Arc::new(TestCallback {
            started: AtomicUsize::new(0),
            failed: AtomicUsize::new(0),
            restarted: AtomicUsize::new(0),
        });

        let mut supervisor = Supervisor::with_callback(SupervisorConfig::default(), callback.clone());

        supervisor.add_process("failing_worker", ChildType::Permanent, || {
            thread::spawn(|| {
                panic!("Intentional failure");
            })
        });

        let supervisor = supervisor.start_monitoring();
        thread::sleep(Duration::from_millis(500));

        // Cast to TestCallback to access fields
        let callback_test = callback.as_ref() as *const dyn EventCallback as *const TestCallback;
        unsafe {
            assert!((*callback_test).started.load(Ordering::Relaxed) > 0);
            assert!((*callback_test).failed.load(Ordering::Relaxed) > 0);
            assert!((*callback_test).restarted.load(Ordering::Relaxed) > 0);
        }

        supervisor.shutdown();
    }
}