dag-executor 0.1.0

A production-ready DAG executor with state management and advanced patterns
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

//! Unit tests for the scheduler's priority queue and state tracking.

use dag_executor::dag::Scheduler;
use dag_executor::state::TaskState;

#[test]
fn serves_highest_priority_first() {
    let mut s = Scheduler::new();
    s.ensure_record("low");
    s.ensure_record("mid");
    s.ensure_record("high");
    s.mark_ready("low", 1);
    s.mark_ready("high", 10);
    s.mark_ready("mid", 5);

    assert_eq!(s.next_ready().as_deref(), Some("high"));
    assert_eq!(s.next_ready().as_deref(), Some("mid"));
    assert_eq!(s.next_ready().as_deref(), Some("low"));
    assert_eq!(s.next_ready(), None);
}

#[test]
fn equal_priority_is_fifo() {
    let mut s = Scheduler::new();
    for id in ["a", "b", "c"] {
        s.ensure_record(id);
        s.mark_ready(id, 0);
    }
    assert_eq!(s.next_ready().as_deref(), Some("a"));
    assert_eq!(s.next_ready().as_deref(), Some("b"));
    assert_eq!(s.next_ready().as_deref(), Some("c"));
}

#[test]
fn skips_stale_ready_entries() {
    let mut s = Scheduler::new();
    s.ensure_record("a");
    s.mark_ready("a", 0);
    // Move it out of Ready; the queued entry is now stale.
    s.transition("a", TaskState::Running);
    assert_eq!(s.next_ready(), None);
}

#[test]
fn tracks_terminal_completion() {
    let mut s = Scheduler::new();
    s.ensure_record("a");
    assert!(!s.all_terminal());
    s.transition("a", TaskState::Ready);
    s.transition("a", TaskState::Running);
    s.transition("a", TaskState::Completed);
    assert!(s.all_terminal());
    assert_eq!(s.count_in(TaskState::Completed), 1);
}

#[test]
fn rejects_illegal_transition() {
    let mut s = Scheduler::new();
    s.ensure_record("a");
    // Pending -> Completed is not a legal direct transition.
    assert!(!s.transition("a", TaskState::Completed));
    assert_eq!(s.state("a"), Some(TaskState::Pending));
}