bcode_workflow_store 0.0.1-alpha.0

Durable workflow persistence for bcode
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

use super::super::tests::{database, run};
use super::*;
#[tokio::test]
async fn graph_larger_than_old_envelope_has_bounded_exact_node_pages() {
    let db = database().await;
    super::super::install(&*db).await.unwrap();
    let mut definition = bcode_workflow::WorkflowBuilder::new(
        "large",
        bcode_workflow::Step::task("seed", |v: u32, _| async move { Ok(v) }),
    )
    .build()
    .unwrap()
    .definition()
    .clone();
    let seed = definition.nodes["seed"].clone();
    definition.nodes.clear();
    definition.entries.clear();
    definition.edges.clear();
    for index in 0..160 {
        let mut node = seed.clone();
        node.id = format!("n{index:04}");
        node.configuration = serde_json::json!({"context":"x".repeat(8192)});
        definition.nodes.insert(node.id.clone(), node);
        if index > 0 {
            definition.edges.push(bcode_workflow::EdgeDefinition {
                from: format!("n{:04}", index - 1),
                to: format!("n{index:04}"),
                kind: bcode_workflow::EdgeKind::Direct,
                transform: None,
            });
        }
    }
    definition.entries = vec!["n0000".into()];
    definition.exits = vec!["n0159".into()];
    assert!(serde_json::to_vec(&definition).unwrap().len() > 1_048_576);
    let mut request = run("large", false);
    let identity =
        bcode_workflow::WorkflowDefinitionIdentity::for_definition("large", &definition).unwrap();
    request.definition_id = identity.definition_id;
    request.definition_version = identity.definition_version;
    assert!(
        super::super::admit_with_policy(
            &*db,
            &request,
            &definition,
            super::super::GraphAdmissionPolicy {
                max_snapshot_bytes: Some(1024)
            }
        )
        .await
        .is_err()
    );
    super::super::admit(&*db, &request, &definition)
        .await
        .unwrap();
    assert_eq!(nodes(&*db, "large", 1, None, 32).await.unwrap().len(), 32);
    let (revision, ready) = super::super::graph::ready_nodes(&*db, "large", None, 32)
        .await
        .unwrap();
    assert_eq!(revision, 1);
    assert_eq!(ready.len(), 1);
    assert_eq!(ready[0].1.id, "n0000");
    assert_eq!(
        node(&*db, "large", 1, "n0159").await.unwrap().unwrap().id,
        "n0159"
    );
}