/home/runner/work/lyquor/lyquor/platform/api/src/subsys.rs

Source
//! Ownership-based lifecycle supervision for long-lived node subsystems.
//!
//! Reusable components expose a dormant configuration whose `spawn` method constructs a
//! [`TaskGroup`](crate::subsys::TaskGroup) and returns a
//! [`SpawnResult`](crate::subsys::SpawnResult) containing the capability handle. After spawning
//! succeeds, the composition root moves the task group into
//! [`Supervisor::register`](crate::subsys::Supervisor::register).
//! [`Supervisor::stop_before`](crate::subsys::Supervisor::stop_before) records lifecycle ordering.
//! Capability flow stays in normal Rust fields and is not coupled to the supervisor graph.
//!
//! Startup remains ordinary imperative code. Register each successfully spawned component
//! immediately, and call [`Supervisor::shutdown`](crate::subsys::Supervisor::shutdown) on a later
//! startup error to unwind everything already registered. Independent startup branches may still
//! be joined explicitly at the composition root, without making subsystem types carry readiness
//! promises.
//!
//! Every graph entry owns one [`TaskGroup`](crate::subsys::TaskGroup), and related workers share
//! clones of it. A task group may create child groups for nested work:
//! parent cancellation propagates to children, while a child may stop independently. At that
//! graph node's turn, its token is cancelled and its whole task-group tree is drained.
//!
//! Shutdown walks the dependency graph in reverse: each node stops after all nodes depending on it
//! have stopped, while independent branches stop concurrently.

use std::{
    future::Future,
    sync::Arc,
    time::{Duration, Instant},
};

use tokio::sync::SetOnce;
use tokio_util::{
    sync::{CancellationToken, DropGuard},
    task::TaskTracker,
};

use crate::log::LogClass;

/// Interval at which the supervisor warns about a subsystem that is slow to stop. The supervisor
/// keeps waiting: this is observability, not a deadline.
const SLOW_WARN_INTERVAL: Duration = Duration::from_secs(10);

/// Cancellation-and-drain scope for a related task group.
///
/// Clone and share a task group among related workers. Use [`Self::child`] only when a nested group
/// needs an independent shutdown boundary. Use [`Self::spawn_cancellable`] when a leaf worker needs
/// the group's shutdown token, and [`Self::spawn_child`] when one worker needs narrower cancellation
/// while remaining tracked by this group. Runtime state that owns a task group should derive group
/// shutdown from it rather than carrying a parallel group-level cancellation token.
///
/// Dropping the last clone cancels the scope without draining it.
#[derive(Clone)]
pub struct TaskGroup {
    shutdown: CancellationToken,
    tasks: TaskTracker,
    _drop_guard: Arc<DropGuard>,
}

lyquor_primitives::debug_struct_name!(TaskGroup);

impl TaskGroup {
    fn new(shutdown: CancellationToken) -> Self {
        let tasks = TaskTracker::new();
        tasks.spawn(shutdown.clone().cancelled_owned());
        tasks.close();
        let drop_guard = Arc::new(shutdown.clone().drop_guard());
        Self {
            shutdown,
            tasks,
            _drop_guard: drop_guard,
        }
    }

    /// Create a nested task-group scope.
    ///
    /// Parent cancellation reaches the child; child cancellation is local. Parent shutdown waits
    /// for the child to drain.
    pub fn child(&self) -> Self {
        let child = Self::new(self.shutdown.child_token());
        let child_tasks = child.tasks.clone();
        self.tasks.spawn(async move {
            child_tasks.wait().await;
        });
        child
    }

    /// Token cancelled when this task group must stop.
    ///
    /// A child token changes signalling, not task ownership.
    pub fn shutdown_token(&self) -> &CancellationToken {
        &self.shutdown
    }

    /// Track one task as part of this group.
    ///
    /// Shutdown waits for all tracked tasks to finish. Worker-owned cleanup should complete inside
    /// the tracked future before it returns.
    ///
    /// Like [`TaskTracker::spawn`], this does not reject inserts after the tracker has been closed.
    /// Runtime request paths that keep a cloned task group should check
    /// [`CancellationToken::is_cancelled`] before spawning new work.
    #[track_caller]
    pub fn spawn<F>(&self, task: F)
    where
        F: Future + Send + 'static,
        F::Output: Send + 'static,
    {
        self.tasks.spawn(task);
    }

    /// Spawn one task that is cancelled with this task group.
    ///
    /// The task receives a clone of this group's shutdown token. Cancelling that token cancels the
    /// entire task group; use [`Self::spawn_child`] when only the spawned task should be cancelled.
    /// The task remains tracked and drained by this group.
    #[track_caller]
    pub fn spawn_cancellable<F, Fut>(&self, run: F)
    where
        F: FnOnce(CancellationToken) -> Fut,
        Fut: Future + Send + 'static,
        Fut::Output: Send + 'static,
    {
        self.spawn(run(self.shutdown.clone()));
    }

    /// Spawn one independently cancellable task owned by this task group.
    ///
    /// The task receives a child token, and the same token is returned to its controller. Cancelling
    /// the child token signals only this task; cancelling the task group also cancels the child token.
    /// The task remains tracked and drained by this group.
    #[must_use = "retain the token to cancel this task independently; use spawn_cancellable otherwise"]
    #[track_caller]
    pub fn spawn_child<F, Fut>(&self, run: F) -> CancellationToken
    where
        F: FnOnce(CancellationToken) -> Fut,
        Fut: Future + Send + 'static,
        Fut::Output: Send + 'static,
    {
        let shutdown = self.shutdown.child_token();
        self.spawn(run(shutdown.clone()));
        shutdown
    }

    /// Wait until this task group has been cancelled and all tracked work and children have
    /// drained. This does not request cancellation.
    pub async fn finished(&self) {
        self.tasks.wait().await;
    }

    /// Cancel this task group and wait for all work spawned through it and its children to finish.
    pub async fn shutdown(self) {
        self.shutdown.cancel();
        self.finished().await;
    }
}

impl Default for TaskGroup {
    fn default() -> Self {
        Self::new(Default::default())
    }
}

/// Standard result of spawning a reusable component.
///
/// `T` is the component's runtime capability. Components without one use `()`. The default error
/// type is [`anyhow::Error`], while components with a typed error can override `E`. A `spawn`
/// implementation must unwind partial work before returning `Err` because task-group ownership has
/// not yet transferred to [`Supervisor::register`].
pub type SpawnResult<T, E = anyhow::Error> = Result<(T, TaskGroup), E>;

/// Identifier of a subsystem or direct task group in one [`Supervisor`]'s lifecycle graph.
///
/// Ids are plain graph indices scoped to the supervisor that issued them. Mixing ids from separate
/// supervisors is unsupported wiring.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct SubsysId(usize);

struct Entry {
    name: &'static str,
    /// Nodes this entry must stop before.
    deps: Vec<usize>,
    task_group: TaskGroup,
}

/// Owner of the node lifecycle graph.
///
/// Registration retains ownership of each task group until ordered shutdown completes or the
/// supervisor is dropped. [`Self::finished`] observes task-group completion without initiating it.
#[derive(Default)]
pub struct Supervisor {
    entries: Vec<Entry>,
}

lyquor_primitives::debug_struct_name!(Supervisor);

impl Supervisor {
    /// Register one task group in the shutdown graph.
    ///
    /// The component creates the task group and spawns its workers through it. After spawning
    /// succeeds, move the returned task group here. It and all its clones must belong to only one
    /// graph entry.
    pub fn register(&mut self, name: &'static str, task_group: TaskGroup) -> SubsysId {
        let id = SubsysId(self.entries.len());
        self.entries.push(Entry {
            name,
            deps: Vec::new(),
            task_group,
        });
        id
    }

    /// Start and register a composition-local task group.
    ///
    /// `run` is called immediately with the new task group. Its returned future is tracked, receives
    /// the shutdown signal through that group, and is drained during ordered shutdown.
    /// Reusable components should expose a dormant configuration whose `spawn` method returns a
    /// [`SpawnResult`], then use [`Self::register`]. This helper is for composition-local task
    /// groups without a separate capability type.
    pub fn spawn<Fut>(&mut self, name: &'static str, run: impl FnOnce(TaskGroup) -> Fut) -> SubsysId
    where
        Fut: Future<Output = ()> + Send + 'static,
    {
        let task_group = TaskGroup::default();
        task_group.spawn(run(task_group.clone()));
        self.register(name, task_group)
    }

    /// Require `first` to stop and drain before `later` begins stopping.
    ///
    /// Use the same primitive for capability dependencies and ordering-only flow constraints. The
    /// graph describes shutdown order only; business handles remain ordinary fields in the
    /// composition root and dormant component configurations.
    ///
    /// # Panics
    /// Panics if either id is out of range for this supervisor or the edge would create a cycle.
    pub fn stop_before(&mut self, first: SubsysId, later: SubsysId) {
        assert!(
            first.0 < self.entries.len() && later.0 < self.entries.len(),
            "stop_before: SubsysId is out of range for this supervisor graph"
        );
        assert_ne!(first, later, "stop_before: a subsystem cannot stop before itself");
        assert!(
            !stops_before(&self.entries, later.0, first.0),
            "stop_before({} before {}) contradicts the existing stop order",
            self.entries[first.0].name,
            self.entries[later.0].name,
        );
        if !self.entries[first.0].deps.contains(&later.0) {
            self.entries[first.0].deps.push(later.0);
        }0
    }

    /// Wait until every registered task group has been cancelled and drained.
    ///
    /// This only observes completion; it does not request cancellation or alter shutdown order.
    pub async fn finished(&self) {
        futures::future::join_all1(self.entries.iter()1.map1(|entry| entry.task_group.finished())).await1;
    }

    /// Stop all entries in reverse dependency layers. Independent branches stop concurrently.
    ///
    /// This consumes the supervisor and is intended to be awaited to completion. It is not a
    /// cancel-safe operation to race in `tokio::select!` and then drop; enforce any process-level
    /// shutdown deadline outside the lifecycle registry.
    pub async fn shutdown(mut self) {
        let started = Instant::now();
        // Empty the supervisor so its Drop safety net does not run after orderly shutdown.
        let entries = std::mem::take(&mut self.entries);

        let mut dependents = vec![Vec::new(); entries.len()];
        for (index, entry) in entries.iter()6.enumerate6() {
            for &dependency3 in &entry.deps {
                dependents[dependency].push(index);
            }
        }

        let stopped6 = Arc::new6(entries.iter()6.map6(|_| SetOnce::new()).collect6::<Vec<_>>());
        let waits6 = entries6.into_iter6().enumerate6().map6(|(index, entry)| {
            let blockers = dependents[index].clone();
            let stopped = stopped.clone();
            async move {
                for blocker3 in blockers {
                    stopped[blocker].wait().await;
                }
                stop_entry(entry.name, entry.task_group).await;
                let _ = stopped[index].set(());
            }
        });
        futures::future::join_all(waits).await;

        tracing::info!(
            elapsed_ms = started.elapsed().as_millis() as u64,
            "all subsystems stopped"
        );
    }
}

/// Safety net for a supervisor dropped without ordered shutdown. Every entry is signalled but not
/// drained; tracked subsystem tasks may continue in the background.
impl Drop for Supervisor {
    fn drop(&mut self) {
        if self.entries.is_empty() {
            return;
        }
        tracing::warn!(
            class = %LogClass::CoreRuntime,
            "supervisor dropped before shutdown completed"
        );
        for entry in &self.entries5 {
            entry.task_group.shutdown.cancel();
        }
    }
}

async fn stop_entry(name: &'static str, task_group: TaskGroup) {
    let started = Instant::now();
    tracing::debug!(subsystem = name, "stopping subsystem");
    warn_if_slow(name, task_group.shutdown()).await;

    tracing::debug!(
        subsystem = name,
        elapsed_ms = started.elapsed().as_millis() as u64,
        "subsystem stopped"
    );
}

async fn warn_if_slow<T>(name: &'static str, operation: impl Future<Output = T>) -> T {
    let started = Instant::now();
    tokio::pin!(operation);
    loop {
        match tokio::time::timeout(SLOW_WARN_INTERVAL, &mut operation).await {
            Ok(output) => break output,
            Err(_) => tracing::warn!(
                subsystem = name,
                elapsed_ms = started.elapsed().as_millis() as u64,
                class = %LogClass::CoreRuntime,
                "subsystem is taking a long time to stop"
            ),
        }
    }
}

fn stops_before(entries: &[Entry], from: usize, to: usize) -> bool {
    let mut seen = vec![false; entries.len()];
    let mut stack = vec![from];
    while let Some(current6) = stack.pop() {
        if current == to {
            return true;
        }
        if std::mem::replace(&mut seen[current], true) {
            continue;
        }
        stack.extend(entries[current].deps.iter().copied());
    }
    false
}

#[cfg(test)]
mod tests {
    use lyquor_test::test;
    use tokio::sync::{mpsc, oneshot};

    use super::*;

    struct TestSubsystem {
        name: &'static str,
        task_group: TaskGroup,
    }

    impl TestSubsystem {
        fn new(
            name: &'static str, events: &mpsc::UnboundedSender<&'static str>, gate: Option<oneshot::Receiver<()>>,
        ) -> (Self, TaskGroup) {
            let task_group = TaskGroup::default();
            let registration_task_group = task_group.clone();
            let worker_task_group = task_group.clone();
            let events = events.clone();
            task_group.spawn(async move {
                worker_task_group.shutdown_token().cancelled().await;
                events.send(name).unwrap();
                if let Some(gate6) = gate {
                    let _ = gate.await;
                }
            });
            (Self { name, task_group }, registration_task_group)
        }

        fn handle(&self) -> &'static str {
            self.name
        }

        fn is_shutdown(&self) -> bool {
            self.task_group.shutdown_token().is_cancelled()
        }
    }

    fn start_test_subsystem(
        sup: &mut Supervisor, name: &'static str, events: &mpsc::UnboundedSender<&'static str>,
        gate: Option<oneshot::Receiver<()>>,
    ) -> (TestSubsystem, SubsysId) {
        let (subsystem, task_group) = TestSubsystem::new(name, events, gate);
        let id = sup.register(name, task_group);
        (subsystem, id)
    }

    #[test(tokio::test)]
    async fn subsystem_handles_remain_ordinary_values() {
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let mut sup = Supervisor::default();
        let (subsystem, _) = start_test_subsystem(&mut sup, "subsystem", &events_tx, None);
        let handle = subsystem.handle();

        assert_eq!(handle, "subsystem");
        assert!(!subsystem.is_shutdown());
        assert!(events_rx.try_recv().is_err());
        sup.shutdown().await;
        assert!(subsystem.is_shutdown());
        assert_eq!(events_rx.recv().await, Some("subsystem"));
    }

    #[test(tokio::test)]
    async fn stop_before_orders_entries() {
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let (release_dependent, dependent_gate) = oneshot::channel();
        let mut sup = Supervisor::default();
        let (_dependency_subsystem, dependency) = start_test_subsystem(&mut sup, "dependency", &events_tx, None);
        let (_dependent_subsystem, dependent) =
            start_test_subsystem(&mut sup, "dependent", &events_tx, Some(dependent_gate));
        sup.stop_before(dependent, dependency);

        let shutdown = tokio::spawn(sup.shutdown());
        assert_eq!(events_rx.recv().await, Some("dependent"));
        assert!(events_rx.try_recv().is_err());
        release_dependent.send(()).unwrap();
        assert_eq!(events_rx.recv().await, Some("dependency"));
        shutdown.await.unwrap();
    }

    #[test(tokio::test)]
    async fn independent_subsystems_stop_concurrently() {
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let (release_x, gate_x) = oneshot::channel();
        let (release_y, gate_y) = oneshot::channel();
        let mut sup = Supervisor::default();
        let (_x, _) = start_test_subsystem(&mut sup, "x", &events_tx, Some(gate_x));
        let (_y, _) = start_test_subsystem(&mut sup, "y", &events_tx, Some(gate_y));

        let shutdown = tokio::spawn(sup.shutdown());
        let first = events_rx.recv().await.unwrap();
        let second = events_rx.recv().await.unwrap();
        assert_ne!(first, second);
        release_x.send(()).unwrap();
        release_y.send(()).unwrap();
        shutdown.await.unwrap();
    }

    #[test(tokio::test)]
    async fn all_dependents_drain_before_dependency() {
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let (release_y, gate_y) = oneshot::channel();
        let (release_z, gate_z) = oneshot::channel();
        let mut sup = Supervisor::default();
        let (_x_subsystem, x) = start_test_subsystem(&mut sup, "x", &events_tx, None);
        let (_y_subsystem, y) = start_test_subsystem(&mut sup, "y", &events_tx, Some(gate_y));
        let (_z_subsystem, z) = start_test_subsystem(&mut sup, "z", &events_tx, Some(gate_z));
        sup.stop_before(y, x);
        sup.stop_before(z, x);

        let shutdown = tokio::spawn(sup.shutdown());
        let first = events_rx.recv().await.unwrap();
        let second = events_rx.recv().await.unwrap();
        assert!(matches!((first, second), ("y", "z") | ("z", "y")));
        assert!(events_rx.try_recv().is_err());
        release_y.send(()).unwrap();
        assert!(events_rx.try_recv().is_err());
        release_z.send(()).unwrap();
        assert_eq!(events_rx.recv().await, Some("x"));
        shutdown.await.unwrap();
    }

    #[test(tokio::test)]
    async fn stop_edges_reject_cycles() {
        let mut sup = Supervisor::default();
        let a = sup.register("a", TaskGroup::default());
        let b = sup.register("b", TaskGroup::default());
        sup.stop_before(b, a);
        let cycle = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| sup.stop_before(a, b)));
        assert!(cycle.is_err());
    }

    #[test(tokio::test)]
    async fn stop_edges_reject_out_of_range_ids() {
        let mut first = Supervisor::default();
        let foreign = first.register("foreign", TaskGroup::default());
        let mut second = Supervisor::default();
        let local = second.register("local", TaskGroup::default());
        let invalid = SubsysId(foreign.0 + 1);
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| second.stop_before(local, invalid)));
        assert!(result.is_err());
    }

    #[test(tokio::test)]
    async fn supervisor_spawn_tasks_are_cancelled_and_drained() {
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let mut sup = Supervisor::default();
        sup.spawn("worker", |task_group| async move {
            task_group.shutdown_token().cancelled().await;
            events_tx.send("stopped").unwrap();
        });

        sup.shutdown().await;
        assert_eq!(events_rx.recv().await, Some("stopped"));
    }

    #[test(tokio::test)]
    async fn direct_task_group_shutdown_awaits_cleanup() {
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let (release_tx, gate) = oneshot::channel();
        let (_subsystem, task_group) = TestSubsystem::new("subsystem", &events_tx, Some(gate));

        let shutdown = tokio::spawn(task_group.shutdown());
        assert_eq!(events_rx.recv().await, Some("subsystem"));
        assert!(!shutdown.is_finished());
        release_tx.send(()).unwrap();
        shutdown.await.unwrap();
    }

    #[test(tokio::test)]
    async fn child_shutdown_does_not_cancel_parent() {
        let parent = TaskGroup::default();
        let child = parent.child();
        let worker_task_group = child.clone();
        let (stopped_tx, stopped_rx) = oneshot::channel();
        child.spawn(async move {
            worker_task_group.shutdown_token().cancelled().await;
            stopped_tx.send(()).unwrap();
        });

        child.shutdown().await;
        stopped_rx.await.unwrap();
        assert!(!parent.shutdown_token().is_cancelled());
        parent.shutdown().await;
    }

    #[test(tokio::test)]
    async fn spawned_child_is_independently_cancellable_and_group_tracked() {
        let task_group = TaskGroup::default();
        let (child_stopped_tx, child_stopped_rx) = oneshot::channel();
        let (release_child_tx, release_child_rx) = oneshot::channel();
        let child_shutdown = task_group.spawn_child(|shutdown| async move {
            shutdown.cancelled().await;
            child_stopped_tx.send(()).unwrap();
            let _ = release_child_rx.await;
        });

        child_shutdown.cancel();
        child_stopped_rx.await.unwrap();
        assert!(!task_group.shutdown_token().is_cancelled());

        let (parent_stopped_tx, parent_stopped_rx) = oneshot::channel();
        let parent_child_shutdown = task_group.spawn_child(|shutdown| async move {
            shutdown.cancelled().await;
            parent_stopped_tx.send(()).unwrap();
        });
        let shutdown = tokio::spawn(task_group.shutdown());
        parent_stopped_rx.await.unwrap();
        assert!(parent_child_shutdown.is_cancelled());
        assert!(!shutdown.is_finished());

        release_child_tx.send(()).unwrap();
        shutdown.await.unwrap();
    }

    #[test(tokio::test)]
    async fn spawned_cancellable_uses_task_group_token_and_is_tracked() {
        let task_group = TaskGroup::default();
        let (cancelled_tx, cancelled_rx) = oneshot::channel();
        let (release_tx, release_rx) = oneshot::channel();
        task_group.spawn_cancellable(|shutdown| async move {
            shutdown.cancel();
            cancelled_tx.send(()).unwrap();
            let _ = release_rx.await;
        });

        cancelled_rx.await.unwrap();
        assert!(task_group.shutdown_token().is_cancelled());

        let shutdown = tokio::spawn(task_group.shutdown());
        assert!(!shutdown.is_finished());
        release_tx.send(()).unwrap();
        shutdown.await.unwrap();
    }

    #[test(tokio::test)]
    async fn dropping_last_clone_cancels_task_group() {
        let task_group = TaskGroup::default();
        let clone = task_group.clone();
        let shutdown = task_group.shutdown_token().clone();

        drop(task_group);
        assert!(!shutdown.is_cancelled());
        drop(clone);
        shutdown.cancelled().await;
    }

    #[test(tokio::test)]
    async fn dropping_child_cancels_only_child() {
        let parent = TaskGroup::default();
        let child = parent.child();
        let child_shutdown = child.shutdown_token().clone();

        drop(child);
        child_shutdown.cancelled().await;
        assert!(!parent.shutdown_token().is_cancelled());
        parent.shutdown().await;
    }

    #[test(tokio::test)]
    async fn parent_shutdown_cancels_and_drains_child() {
        let parent = TaskGroup::default();
        let child = parent.child();
        let worker_task_group = child.clone();
        let (stopped_tx, stopped_rx) = oneshot::channel();
        let (release_tx, release_rx) = oneshot::channel();
        child.spawn(async move {
            worker_task_group.shutdown_token().cancelled().await;
            stopped_tx.send(()).unwrap();
            let _ = release_rx.await;
        });

        let shutdown = tokio::spawn(parent.shutdown());
        stopped_rx.await.unwrap();
        assert!(!shutdown.is_finished());
        release_tx.send(()).unwrap();
        shutdown.await.unwrap();
    }

    #[test(tokio::test)]
    async fn finished_waits_for_all_registered_task_groups_without_cancelling_them() {
        let mut sup = Supervisor::default();
        let first = TaskGroup::default();
        let first_shutdown = first.shutdown_token().clone();
        let (release_first_tx, release_first_rx) = oneshot::channel();
        let (first_drained_tx, first_drained_rx) = oneshot::channel();
        first.spawn(async move {
            first_shutdown.cancelled().await;
            let _ = release_first_rx.await;
            first_drained_tx.send(()).unwrap();
        });
        let first_shutdown = first.shutdown_token().clone();
        sup.register("first", first);

        let second = TaskGroup::default();
        let second_shutdown = second.shutdown_token().clone();
        sup.register("second", second);

        let finished = tokio::spawn(async move {
            sup.finished().await;
            sup
        });
        tokio::task::yield_now().await;
        assert!(!first_shutdown.is_cancelled());
        assert!(!second_shutdown.is_cancelled());
        assert!(!finished.is_finished());

        first_shutdown.cancel();
        release_first_tx.send(()).unwrap();
        first_drained_rx.await.unwrap();
        assert!(!finished.is_finished());

        second_shutdown.cancel();
        finished.await.unwrap().shutdown().await;
    }

    #[test(tokio::test)]
    async fn dropping_supervisor_signals_subsystem_tasks() {
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let mut sup = Supervisor::default();
        let (_subsystem, _) = start_test_subsystem(&mut sup, "subsystem", &events_tx, None);

        drop(sup);
        assert_eq!(events_rx.recv().await, Some("subsystem"));
    }

    #[test(tokio::test)]
    async fn dropping_supervisor_signals_spawned_tasks() {
        let (stopped_tx, stopped_rx) = oneshot::channel();
        let mut sup = Supervisor::default();
        sup.spawn("worker", |task_group| async move {
            task_group.shutdown_token().cancelled().await;
            stopped_tx.send(()).unwrap();
        });

        drop(sup);
        stopped_rx.await.unwrap();
    }
}

Coverage Report

Created: 2026-10-05 06:05