cymule-activation-timer 0.1.4

Durable logical timer source for Cymule
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

//! Deterministic timer due-time, redelivery, and acknowledgement tests.

use std::collections::{BTreeMap, BTreeSet};

use cymule_activation_timer::{Clock, SqliteTimerDriver};
use cymule_core::{ArtifactRef, Machine, ROOT_SCOPE_ID};
use cymule_durable::{
    Continuation, ContinuationStatus, DurableState, FrameState, ParkedWaitIndex, WaitCondition,
    WaitKind, WaitSourceDriver, WaitState,
};
use serde_json::json;

#[derive(Clone, Copy)]
struct ManualClock(u64);

impl Clock for ManualClock {
    fn now_unix_ms(&self) -> cymule_durable::DurableResult<u64> {
        Ok(self.0)
    }
}

fn index() -> ParkedWaitIndex {
    let wait_id = "wait:timer";
    let mut state = DurableState::new(Machine::new().snapshot());
    state.continuations.insert(
        "run:timer".to_owned(),
        Continuation {
            run_id: "run:timer".to_owned(),
            plan_id: "sha256:plan".to_owned(),
            binding_context: "binding:test".to_owned(),
            frames: vec![FrameState {
                definition_id: "main".to_owned(),
                invocation_id: "main".to_owned(),
                input: ArtifactRef {
                    artifact_id: format!("sha256:{}", "0".repeat(64)),
                    kind: "test/input".to_owned(),
                },
                region_path: Vec::new(),
                next_step: 0,
                locals: BTreeMap::new(),
            }],
            state: None,
            wait_set: BTreeSet::from([wait_id.to_owned()]),
            scope_stack: vec![ROOT_SCOPE_ID.to_owned()],
            effect_obligations: BTreeSet::new(),
            authority_leases: BTreeSet::new(),
            budget: BTreeMap::new(),
            causal_frontier: BTreeSet::new(),
            epoch: 0,
            status: ContinuationStatus::Waiting,
        },
    );
    state.waits.insert(
        wait_id.to_owned(),
        WaitCondition {
            wait_id: wait_id.to_owned(),
            run_id: "run:timer".to_owned(),
            kind: WaitKind::Timer {
                timer_id: "timer:one".to_owned(),
            },
            consume_once: false,
            state: WaitState::Pending,
            result: None,
        },
    );
    ParkedWaitIndex::rebuild(&state).expect("index rebuilds")
}

#[test]
fn due_timer_redelivers_until_acknowledged() {
    let mut driver =
        SqliteTimerDriver::in_memory_with_clock(ManualClock(100)).expect("driver opens");
    driver
        .schedule("activation:one", "timer:one", 100, &json!({"due": true}))
        .expect("timer schedules");
    driver
        .schedule("activation:one", "timer:one", 100, &json!({"due": true}))
        .expect("schedule replay is idempotent");
    let first = driver
        .receive(&index(), 1)
        .expect("timer receives")
        .expect("delivery exists");
    assert_eq!(first.activation_id, "activation:one");
    assert_eq!(
        driver.receive(&index(), 1).expect("redelivers"),
        Some(first)
    );
    driver.acknowledge("activation:one").expect("acknowledges");
    driver
        .acknowledge("activation:one")
        .expect("ack replay succeeds");
    driver
        .schedule("activation:one", "timer:one", 100, &json!({"due": true}))
        .expect("schedule replay after ack succeeds");
    for (timer_id, due, value) in [
        ("timer:other", 100, json!({"due": true})),
        ("timer:one", 101, json!({"due": true})),
        ("timer:one", 100, json!({"due": false})),
    ] {
        assert!(matches!(
            driver.schedule("activation:one", timer_id, due, &value),
            Err(cymule_durable::DurableError::Conflict { .. })
        ));
    }
    assert!(driver.receive(&index(), 1).expect("polls").is_none());
}

#[test]
fn timer_before_due_or_without_a_wait_stays_pending() {
    let mut driver =
        SqliteTimerDriver::in_memory_with_clock(ManualClock(99)).expect("driver opens");
    driver
        .schedule("activation:one", "timer:one", 100, &json!(null))
        .expect("timer schedules");
    assert!(driver.receive(&index(), 1).expect("polls").is_none());
}