task-supervisor 0.5.2

Supervisor for long-lived tokio tasks: restarts with backoff, graceful shutdown, runtime control
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
103
104
105
106
107
108

mod common;

use std::sync::atomic::Ordering;
use std::time::Duration;

use task_supervisor::{SupervisorBuilder, SupervisorError, TaskStatus};
use tokio::time::pause;

use common::{sleep_ms, Cooperative, Failing};

fn builder() -> SupervisorBuilder {
    SupervisorBuilder::new()
        .with_restart_limit(0, Duration::from_secs(60))
        .with_base_restart_delay(Duration::from_millis(10))
}

#[tokio::test]
async fn supervisor_shuts_down_when_threshold_reached() {
    pause();
    let steady = Cooperative::default();
    let handle = builder()
        .with_dead_tasks_threshold(0.5)
        .with_task("steady", steady.clone())
        .with_task("flaky", Failing::default())
        .spawn();

    let err = handle.wait().await.unwrap_err();
    assert!(
        matches!(
            err,
            SupervisorError::TooManyDeadTasks {
                dead: 1,
                total: 2,
                ..
            }
        ),
        "{err}"
    );
    assert_eq!(
        err.to_string(),
        "too many dead tasks: 1/2 (50%) >= threshold 50%"
    );
    assert!(
        steady.stopped.load(Ordering::SeqCst),
        "surviving tasks are stopped gracefully"
    );
}

#[tokio::test]
async fn supervisor_keeps_running_below_threshold() {
    pause();
    let handle = builder()
        .with_dead_tasks_threshold(0.6)
        .with_task("steady", Cooperative::default())
        .with_task("flaky", Failing::default())
        .spawn();

    sleep_ms(100).await;
    assert_eq!(
        handle.task_status("steady").await.unwrap(),
        TaskStatus::Running
    );
    assert_eq!(handle.task_status("flaky").await.unwrap(), TaskStatus::Dead);
    assert!(handle.shutdown().await.is_ok());
}

/// Regression: `0.0` means "any dead task", not "shut down at the first event".
#[tokio::test]
async fn threshold_zero_waits_for_a_death() {
    pause();
    let handle = builder()
        .with_dead_tasks_threshold(0.0)
        .with_task("steady", Cooperative::default())
        .spawn();

    sleep_ms(100).await;
    assert_eq!(
        handle.task_status("steady").await.unwrap(),
        TaskStatus::Running
    );

    handle.kill_task("steady").await.unwrap();
    assert!(matches!(
        handle.wait().await,
        Err(SupervisorError::TooManyDeadTasks {
            dead: 1,
            total: 1,
            ..
        })
    ));
}

#[tokio::test]
async fn threshold_zero_ignores_restarts() {
    pause();
    let handle = builder()
        .with_dead_tasks_threshold(0.0)
        .with_task("steady", Cooperative::default())
        .spawn();

    handle.restart_task("steady").await.unwrap();
    sleep_ms(100).await;
    assert_eq!(
        handle.task_status("steady").await.unwrap(),
        TaskStatus::Running
    );
    assert!(handle.shutdown().await.is_ok());
}