bcode_server 0.0.1-alpha.0

Local bcode daemon runtime
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

use super::{Category, owner_category};
use bcode_workflow_store::WorkflowStoreError;

#[test]
fn classification_never_retains_private_owner_messages() {
    assert_eq!(
        owner_category(&WorkflowStoreError::OwnerAccessDeferred),
        Category::OwnerAccessDeferred
    );
    let category = owner_category(&WorkflowStoreError::InvalidData(
        "secret-token private-path provider-wire".into(),
    ));
    assert_eq!(category, Category::OwnerError);
    assert_eq!(
        serde_json::to_value(category).unwrap(),
        serde_json::json!("owner_error")
    );
}

mod fixture;

#[tokio::test]
async fn dispatch_failure_is_durable_without_terminalizing_or_retrying() {
    let (root, state, db, owner, request) = Box::pin(fixture::prepare()).await;
    let before = bcode_workflow_store::representation::attempts::owner_boundary::prepared(
        &db,
        &request.activation.run_id,
        &request.dispatch_identity,
    )
    .await
    .unwrap()
    .expect("prepared before failure");
    assert!(
        super::super::dispatch_and_record(&state, &owner, &request, &fixture::RejectingOwner)
            .await
            .is_err()
    );
    let rows = bcode_workflow_store::representation::admission_observation::inspect(
        &db,
        &request.activation.run_id,
        None,
        10,
    )
    .await
    .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].category, Category::OwnerError);
    assert_eq!(rows[0].dispatch_identity, request.dispatch_identity);
    let boundary = bcode_workflow_store::representation::attempts::owner_boundary::receipt(
        &db,
        &request.dispatch_identity,
    )
    .await
    .unwrap();
    assert!(boundary.is_none());
    assert!(
        bcode_workflow_store::representation::attempts::owner_boundary::terminal_outcome(
            &db,
            &request.activation.run_id,
            &request.dispatch_identity
        )
        .await
        .unwrap()
        .is_none()
    );
    let after = bcode_workflow_store::representation::attempts::owner_boundary::prepared(
        &db,
        &request.activation.run_id,
        &request.dispatch_identity,
    )
    .await
    .unwrap()
    .expect("preparation retained");
    assert_eq!(before, after);
    drop(db);
    drop(state);
    drop(root);
}

#[tokio::test]
async fn stale_authority_cannot_record_failure_or_change_preparation() {
    let (root, state, db, mut owner, request) = Box::pin(fixture::prepare()).await;
    let before = bcode_workflow_store::representation::attempts::owner_boundary::prepared(
        &db,
        &request.activation.run_id,
        &request.dispatch_identity,
    )
    .await
    .unwrap();
    owner.generation += 1;
    assert!(
        super::record(
            &state,
            &owner,
            &request,
            bcode_workflow_store::representation::admission_observation::Phase::OwnerDispatch,
            Category::OwnerError,
        )
        .await
        .is_err()
    );
    let observations = bcode_workflow_store::representation::admission_observation::inspect(
        &db,
        &request.activation.run_id,
        None,
        10,
    )
    .await
    .unwrap();
    assert!(observations.is_empty());
    let after = bcode_workflow_store::representation::attempts::owner_boundary::prepared(
        &db,
        &request.activation.run_id,
        &request.dispatch_identity,
    )
    .await
    .unwrap();
    assert_eq!(before, after);
    drop(db);
    drop(state);
    drop(root);
}