Dynamic task management

Use Supervisor::serve when tasks are discovered after the runtime starts. It returns a handle for registration, inspection, stopping, and joined shutdown.

application ──► serve ──► SupervisorHandle
                                ├── add ───────► registry ──► task attempts
                                ├── remove ────► claim a stop and return
                                ├── cancel ────► wait for cleanup
                                └── shutdown ──► close admission and wait

add confirms registry admission. It does not confirm that an attempt started or finished. list reads registry membership. is_alive reads physical attempt activity directly. Neither query depends on lifecycle-event delivery.

remove returns after it claims a stop, before registered-task cleanup finishes. cancel waits for bounded logical cleanup. Here, cancel follows an earlier remove. It joins that stop or observes completed cleanup. It returns false in either case because it did not create the original claim.

Expect worker ticks, three Registered: snapshots, and cancellation messages. The example requests shutdown itself and exits after a few seconds.

Run with cargo run --example dynamic_tasks.

Complete program

use std::time::Duration;

use taskvisor::prelude::*;

fn make_worker(name: &'static str) -> TaskSpec {
    let task: TaskRef = TaskFn::arc(move |ctx| async move {
        let mut tick = 0u32;
        loop {
            match ctx
                .run_until_cancelled(tokio::time::sleep(Duration::from_millis(300)))
                .await
            {
                Ok(()) => {
                    tick += 1;
                    println!("  [{name}] tick #{tick}");
                }
                Err(canceled) => {
                    println!("  [{name}] stopped at tick #{tick}");
                    return Err(canceled);
                }
            }
        }
    });
    TaskSpec::restartable(name, task)
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let supervisor = Supervisor::new(SupervisorConfig::default(), vec![]);

    // serve() starts the runtime without OS signal handlers and returns its management handle.
    let handle = supervisor.serve()?;

    // Add workers dynamically
    println!("Adding worker-a and worker-b...");
    let id_a = handle.add(make_worker("worker-a")).await?;
    let id_b = handle.add(make_worker("worker-b")).await?;

    // Demo pacing only: add() has already confirmed registration.
    tokio::time::sleep(Duration::from_secs(1)).await;
    println!("Registered: {:?}", handle.list().await);

    // remove() claims the stop but returns before registered-task cleanup ends.
    println!("\nRemoving worker-a...");
    let removed = handle.remove(id_a).await?;
    println!("worker-a removal claimed: {removed}");

    // Join the same removal before reading authoritative registry state.
    // This returns false because remove() already claimed the stop (or cleanup finished first).
    let second_claim = handle.cancel(id_a).await?;
    println!("worker-a second cancellation claimed: {second_claim}");
    println!("Registered: {:?}", handle.list().await);

    // Add worker-c
    println!("\nAdding worker-c...");
    handle.add(make_worker("worker-c")).await?;
    // Demo pacing only: let the terminal show a worker-c tick.
    tokio::time::sleep(Duration::from_millis(500)).await;

    // Cancel worker-b
    println!("Cancelling worker-b...");
    let cancelled = handle.cancel(id_b).await?;
    println!("worker-b cancelled: {cancelled}");
    // This direct query reads physical attempt activity. It does not consume lifecycle events.
    println!(
        "worker-b physically active: {}",
        handle.is_alive("worker-b").await
    );

    // Demo pacing only: let worker-c keep ticking before the final snapshot.
    tokio::time::sleep(Duration::from_millis(500)).await;
    println!("\nRegistered: {:?}", handle.list().await);

    // Graceful shutdown (consumes the handle)
    println!("\nShutting down...");
    handle.shutdown().await?;
    Ok(())
}