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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

//! Per-attempt timeout integration tests.

mod common;

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

use common::*;
use taskvisor::BackoffSource;
use taskvisor::prelude::*;

async fn run_to_exhaustion(spec: TaskSpec) -> Arc<EventCollector> {
    let (supervisor, collector) = supervisor_with_collector(SupervisorConfig::default());
    with_timeout(10, supervisor.run(vec![spec]))
        .await
        .expect("run() should return Ok");
    collector
        .wait_for(EventKind::ActorExhausted, Duration::from_secs(2))
        .await
        .expect("ActorExhausted was not observed");
    collector
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn per_attempt_timeout_emits_timeout_hit_before_task_failed_then_retries() {
    let task = TaskFn::arc("slow", |_ctx: TaskContext| async move {
        tokio::time::sleep(Duration::from_secs(3600)).await;
        Ok(())
    });
    let spec = TaskSpec::restartable(task)
        .with_timeout(Duration::from_millis(50))
        .with_backoff(fast_backoff())
        .with_max_retries(NonZeroU32::new(1).unwrap());
    let collector = run_to_exhaustion(spec).await;

    assert_eq!(collector.count(EventKind::TimeoutHit), 2);
    assert_eq!(collector.count(EventKind::TaskFailed), 2);
    assert_eq!(collector.count(EventKind::BackoffScheduled), 1);
    assert_eq!(collector.count(EventKind::ActorExhausted), 1);

    for attempt in 1..=2u32 {
        let hit = collector
            .find_all(EventKind::TimeoutHit)
            .into_iter()
            .find(|e| e.attempt == Some(attempt))
            .unwrap();
        let failed = collector
            .find_all(EventKind::TaskFailed)
            .into_iter()
            .find(|e| e.attempt == Some(attempt))
            .unwrap();
        assert!(hit.seq < failed.seq, "TimeoutHit must precede TaskFailed");
        assert!(failed.reason.as_deref().unwrap().contains("timed out"));
        assert_eq!(failed.exit_code, None);
    }

    let reason = collector
        .find(EventKind::ActorExhausted)
        .unwrap()
        .reason
        .unwrap();
    assert!(reason.contains("max_retries_exceeded") && reason.contains("(1/1)"));
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn timeout_then_success_unlimited_retries_exhausts_on_success() {
    let n = Arc::new(AtomicU32::new(0));
    let nc = n.clone();
    let task = TaskFn::arc("slow-then-ok", move |_ctx: TaskContext| {
        let nc = nc.clone();
        async move {
            let c = nc.fetch_add(1, Ordering::SeqCst);
            if c == 0 {
                tokio::time::sleep(Duration::from_secs(3600)).await;
            }
            Ok(())
        }
    });
    let spec = TaskSpec::restartable(task)
        .with_timeout(Duration::from_millis(50))
        .with_backoff(fast_backoff());
    let collector = run_to_exhaustion(spec).await;

    assert_eq!(collector.count(EventKind::TimeoutHit), 1);
    assert_eq!(collector.count(EventKind::TaskFailed), 1);
    assert_eq!(collector.count(EventKind::BackoffScheduled), 1);
    assert_eq!(collector.count(EventKind::TaskStarting), 2);
    assert_eq!(collector.count(EventKind::TaskStopped), 1);
    assert_eq!(collector.count(EventKind::ActorExhausted), 1);
    assert_eq!(collector.count(EventKind::ActorDead), 0);

    assert_eq!(
        collector
            .find(EventKind::BackoffScheduled)
            .unwrap()
            .backoff_source,
        Some(BackoffSource::Failure)
    );
    assert_eq!(
        collector
            .find(EventKind::ActorExhausted)
            .unwrap()
            .reason
            .as_deref(),
        Some("policy_exhausted_success")
    );
}

#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn zero_timeout_means_no_timeout_task_runs_to_completion() {
    let task = TaskFn::arc("zero-to", |_ctx: TaskContext| async move {
        tokio::time::sleep(Duration::from_millis(30)).await;
        Ok(())
    });
    let spec = TaskSpec::once(task).with_timeout(Duration::ZERO);
    let collector = run_to_exhaustion(spec).await;

    assert_eq!(collector.count(EventKind::TimeoutHit), 0);
    assert_eq!(collector.count(EventKind::TaskStopped), 1);
    assert_eq!(collector.count(EventKind::TaskFailed), 0);
    assert_eq!(
        collector
            .find(EventKind::ActorExhausted)
            .unwrap()
            .reason
            .as_deref(),
        Some("policy_exhausted_success")
    );
}