lash-core 0.1.0-alpha.88

Sans-IO turn machine and runtime kernel for the lash agent runtime.
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

use std::collections::BTreeMap;

use super::*;

pub(super) async fn live_reference_summary_tracks_non_terminal_reference_counts(
    registry: Arc<dyn ProcessRegistry>,
) {
    assert!(
        registry
            .live_reference_summary()
            .await
            .expect("empty summary")
            .is_empty(),
        "a fresh registry has no live references"
    );

    let definition_a = serde_json::json!({ "module": "alpha", "process": "main" });
    let definition_b = serde_json::json!({ "module": "beta", "process": "main" });
    let env_a = ProcessExecutionEnvRef::new("process-env:alpha");
    let env_b = ProcessExecutionEnvRef::new("process-env:beta");
    for (process_id, definition, env_ref) in [
        ("proc-ref-a1", definition_a.clone(), env_a.clone()),
        ("proc-ref-a2", definition_a.clone(), env_a.clone()),
        ("proc-ref-b", definition_b.clone(), env_b.clone()),
    ] {
        registry
            .register_process(
                ProcessRegistration::new(
                    process_id,
                    ProcessInput::Engine {
                        kind: "reference-test".to_string(),
                        payload: serde_json::Value::Null,
                    },
                    RecoveryDisposition::Rerunnable,
                    ProcessProvenance::host(),
                )
                .with_identity(
                    ProcessIdentity::new("reference-test").with_definition(Some(definition)),
                )
                .with_execution_env_ref(Some(env_ref)),
            )
            .await
            .expect("register reference process");
    }

    let counts = reference_counts(registry.live_reference_summary().await.expect("summary"));
    assert_eq!(
        counts.get(&(key(&definition_a), env_a.to_string())),
        Some(&2)
    );
    assert_eq!(
        counts.get(&(key(&definition_b), env_b.to_string())),
        Some(&1)
    );

    registry
        .complete_process(
            "proc-ref-a1",
            ProcessAwaitOutput::Success {
                value: serde_json::Value::Null,
                control: None,
            },
            crate::ProcessCompletionAuthority::workflow_key("proc-ref-a1"),
        )
        .await
        .expect("complete one alpha process");
    let counts = reference_counts(registry.live_reference_summary().await.expect("summary"));
    assert_eq!(
        counts.get(&(key(&definition_a), env_a.to_string())),
        Some(&1)
    );
    assert_eq!(
        counts.get(&(key(&definition_b), env_b.to_string())),
        Some(&1)
    );

    for process_id in ["proc-ref-a2", "proc-ref-b"] {
        registry
            .complete_process(
                process_id,
                ProcessAwaitOutput::Success {
                    value: serde_json::Value::Null,
                    control: None,
                },
                crate::ProcessCompletionAuthority::workflow_key(process_id),
            )
            .await
            .expect("complete remaining process");
    }
    assert!(
        registry
            .live_reference_summary()
            .await
            .expect("drained summary")
            .is_empty(),
        "summary is empty when all processes using the references are terminal"
    );
}

fn reference_counts(
    summaries: Vec<ProcessLiveReferenceSummary>,
) -> BTreeMap<(String, String), usize> {
    summaries
        .into_iter()
        .map(|summary| {
            (
                (
                    summary.definition.as_ref().map(key).unwrap_or_default(),
                    summary
                        .env_ref
                        .as_ref()
                        .map(ToString::to_string)
                        .unwrap_or_default(),
                ),
                summary.process_count,
            )
        })
        .collect()
}

fn key(definition: &serde_json::Value) -> String {
    serde_json::to_string(definition).expect("definition serializes")
}