later 0.0.48

Distributed Background jobs manager and runner for Rust
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

use crate::models::Stage;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) struct JobTiming {
    pub(crate) total_ms: i64,
    pub(crate) processing_ms: i64,
    pub(crate) waiting_ms: i64,
}

pub(crate) fn calculate(
    previous_stages: &[Stage],
    current_stage: &Stage,
    now: chrono::DateTime<chrono::Utc>,
) -> JobTiming {
    let started_at = previous_stages.first().unwrap_or(current_stage).date();
    let finished_at = if current_stage.is_terminal() {
        current_stage.date()
    } else {
        &now
    };
    let total_ms = elapsed_ms(started_at, finished_at);

    let mut processing_ms = 0_i64;
    for (index, stage) in previous_stages.iter().enumerate() {
        if let Stage::Running(running) = stage {
            let next_stage = previous_stages.get(index + 1).unwrap_or(current_stage);
            processing_ms =
                processing_ms.saturating_add(elapsed_ms(&running.date, next_stage.date()));
        }
    }
    if let Stage::Running(running) = current_stage {
        processing_ms = processing_ms.saturating_add(elapsed_ms(&running.date, &now));
    }
    processing_ms = processing_ms.min(total_ms);

    JobTiming {
        total_ms,
        processing_ms,
        waiting_ms: total_ms.saturating_sub(processing_ms),
    }
}

fn elapsed_ms(
    started_at: &chrono::DateTime<chrono::Utc>,
    finished_at: &chrono::DateTime<chrono::Utc>,
) -> i64 {
    finished_at
        .signed_duration_since(*started_at)
        .num_milliseconds()
        .max(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::models::{EnqueuedStage, RequeuedStage, RunningStage, SuccessStage};

    fn at(second: i64) -> chrono::DateTime<chrono::Utc> {
        chrono::DateTime::from_timestamp(second, 0).unwrap_or_default()
    }

    #[test]
    fn separates_processing_from_queue_and_retry_waits() {
        let previous_stages = vec![
            Stage::Enqueued(EnqueuedStage { date: at(0) }),
            Stage::Running(RunningStage { date: at(2) }),
            Stage::Requeued(RequeuedStage {
                date: at(5),
                requeue_count: 1,
                not_before: Some(at(8)),
            }),
            Stage::Enqueued(EnqueuedStage { date: at(8) }),
            Stage::Running(RunningStage { date: at(10) }),
        ];
        let current_stage = Stage::Success(SuccessStage { date: at(14) });

        assert_eq!(
            calculate(&previous_stages, &current_stage, at(30)),
            JobTiming {
                total_ms: 14_000,
                processing_ms: 7_000,
                waiting_ms: 7_000,
            }
        );
    }

    #[test]
    fn active_job_duration_uses_now_and_never_becomes_negative() {
        assert_eq!(
            calculate(&[], &Stage::Running(RunningStage { date: at(20) }), at(25),),
            JobTiming {
                total_ms: 5_000,
                processing_ms: 5_000,
                waiting_ms: 0,
            }
        );
        assert_eq!(
            calculate(
                &[],
                &Stage::Enqueued(EnqueuedStage { date: at(20) }),
                at(10),
            ),
            JobTiming {
                total_ms: 0,
                processing_ms: 0,
                waiting_ms: 0,
            }
        );
    }
}