taskvisor 0.7.0

In-process Tokio task supervisor with retries, reliable outcomes, and per-key queue/replace/reject admission
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
94
95
96
97
98
99
100
101
102

//! # Outcomes: wait for the final result
//!
//! `add_and_watch` returns a `TaskWaiter`.
//! The waiter resolves after all allowed attempts end, the managed runner is joined, and registry membership is removed.
//!
//! Taskvisor has two result paths:
//!
//! | Path             | Use                          | Delivery                   |
//! |------------------|------------------------------|----------------------------|
//! | lifecycle events | logs, metrics, live progress | bounded and best-effort    |
//! | `TaskOutcome`    | final business decision      | dedicated terminal channel |
//!
//! This example handles successful, retry-exhausted, timed-out, and canceled tasks.
//! Other outcomes cover fatal errors, force-abort, task-runner panic, and controller rejection.
//! `TaskWaiter::wait` can still return an error if the runtime closes the terminal channel unexpectedly.
//!
//! Run with `cargo run --example outcomes`.

use std::num::NonZeroU32;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;

use taskvisor::prelude::*;
use tokio::sync::Notify;

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

    // 1) A one-shot job that succeeds -> Completed.
    println!("=== Completed ===");
    let job: TaskRef = TaskFn::arc("import", |_ctx| async {
        tokio::time::sleep(Duration::from_millis(100)).await;
        Ok(())
    });
    let (_id, waiter) = handle.add_and_watch(TaskSpec::once(job)).await?;
    println!("  import -> {:?}\n", waiter.wait().await?);

    // 2) A task that always fails, with a bounded retry budget -> Failed.
    //    Its reason/exit_code are identical to the typed TaskFinished event.
    println!("=== Failed (retries exhausted) ===");
    let attempts = Arc::new(AtomicU32::new(0));
    let flaky: TaskRef = TaskFn::arc("sync", move |_ctx| {
        let attempts = Arc::clone(&attempts);
        async move {
            let n = attempts.fetch_add(1, Ordering::Relaxed) + 1;
            println!("  sync attempt #{n} failing...");
            Err(TaskError::fail("upstream 503").with_exit_code(75))
        }
    });
    let spec = TaskSpec::restartable(flaky)
        .with_backoff(BackoffPolicy::constant(Duration::from_millis(20)))
        .with_max_retries(NonZeroU32::new(2).unwrap());
    match handle.add_and_watch(spec).await?.1.wait().await? {
        TaskOutcome::Failed {
            reason, exit_code, ..
        } => {
            println!("  sync -> Failed: {reason} (exit_code={exit_code:?})\n");
        }
        other => println!("  sync -> {other:?}\n"),
    }

    // 3) A one-shot task that exceeds its per-attempt deadline -> Failed.
    println!("=== Failed (attempt timed out) ===");
    let slow: TaskRef = TaskFn::arc("slow-report", |_ctx| async {
        tokio::time::sleep(Duration::from_secs(1)).await;
        Ok(())
    });
    let timed = TaskSpec::once(slow).with_timeout(Duration::from_millis(20));
    match handle.add_and_watch(timed).await?.1.wait().await? {
        TaskOutcome::Failed { reason, .. } => {
            println!("  slow-report -> Failed: {reason}\n");
        }
        other => println!("  slow-report -> {other:?}\n"),
    }

    // 4) A long-running worker we cancel -> Canceled.
    println!("=== Canceled ===");
    let started = Arc::new(Notify::new());
    let worker: TaskRef = TaskFn::arc("worker", {
        let started = Arc::clone(&started);
        move |ctx| {
            let started = Arc::clone(&started);
            async move {
                started.notify_one();
                ctx.cancelled().await;
                Err(TaskError::Canceled)
            }
        }
    });
    let (id, waiter) = handle.add_and_watch(TaskSpec::restartable(worker)).await?;
    // The task body, rather than a timer, confirms that the worker started.
    started.notified().await;
    println!("  cancelling worker...");
    handle.cancel(id).await?;
    println!("  worker -> {:?}\n", waiter.wait().await?);

    handle.shutdown().await?;
    Ok(())
}