taskvisor 0.6.0

Task supervisor for Tokio: restarts background tasks on failure with exponential backoff and jitter, graceful shutdown, and lifecycle events
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

//! # Dynamic task management
//!
//! Use `Supervisor::serve` when long-lived tasks are discovered at runtime.
//! Examples include tenant workers, plugins, and connections that own a resident background loop.
//! Taskvisor is not a per-request job executor.
//!
//! This example shows how to:
//!
//! - add a task and receive its `TaskId`;
//! - inspect the current registry snapshot;
//! - cancel or remove a task by identity;
//! - shut down the supervisor explicitly.
//!
//! `Supervisor::run` is the simpler choice when all tasks are known at startup.
//! It waits for the supervisor to finish.
//!
//! Run with `cargo run --example dynamic`.

use std::time::Duration;

use taskvisor::prelude::*;

fn make_worker(name: &'static str) -> TaskSpec {
    let task: TaskRef = TaskFn::arc(name, 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(task)
}

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

    // serve() starts listeners and returns a handle for dynamic management.
    let handle = sup.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?;

    tokio::time::sleep(Duration::from_secs(1)).await;
    println!("Active: {:?}", 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!("Active: {:?}", handle.list().await);

    // Add worker-c
    println!("\nAdding worker-c...");
    handle.add(make_worker("worker-c")).await?;
    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}");
    println!(
        "worker-b alive (best-effort event view): {}",
        handle.is_alive("worker-b").await
    );

    tokio::time::sleep(Duration::from_millis(500)).await;
    println!("\nActive: {:?}", handle.list().await);

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