bpm-engine 0.2.0

Lightweight embeddable BPM runtime for long-running, stateful workflows with tokens, timers, Saga compensation, and crash recovery
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

//! Invariant tests: token lifecycle, exactly-once completion, valid state transitions, single owner.
//! See docs/invariants.md ยง1.

use bpm_engine::bpm_engine_adapter_memory::MemoryRepo;
use bpm_engine::bpm_engine_core::{InstanceState, ProcessInstance, Token, TokenMode, TokenStatus};
use bpm_engine::bpm_engine_storage::{ProcessInstanceStore, TokenStore};
use std::sync::Arc;

#[tokio::test]
async fn token_completes_at_most_once() {
    let repo = Arc::new(MemoryRepo::new());
    let instance_id = "inst-1".to_string();
    let token_id = "token-1".to_string();
    let inst = ProcessInstance {
        id: instance_id.clone(),
        process_def_id: "p".into(),
        tenant_id: None,
        tokens: vec![Token {
            id: token_id.clone(),
            node_id: "n1".into(),
            status: TokenStatus::Completed,
            mode: TokenMode::Forward,
            version: 1,
            attempt: 0,
            parallel_group_id: None,
            updated_at: None,
        }],
        variables: std::collections::HashMap::new(),
        state: InstanceState::Running,
        version: 0,
    };
    repo.save(&inst).await.unwrap();
    let claimed = repo
        .claim_token(&instance_id, &token_id, 1)
        .await
        .unwrap_or(false);
    assert!(
        !claimed,
        "completed token must not be claimable (exactly-once completion)"
    );
}

#[tokio::test]
async fn only_one_claim_succeeds_per_token() {
    let repo = Arc::new(MemoryRepo::new());
    let instance_id = "inst-1".to_string();
    let token_id = "token-1".to_string();
    let inst = ProcessInstance {
        id: instance_id.clone(),
        process_def_id: "p".into(),
        tenant_id: None,
        tokens: vec![Token {
            id: token_id.clone(),
            node_id: "n1".into(),
            status: TokenStatus::Ready,
            mode: TokenMode::Forward,
            version: 0,
            attempt: 0,
            parallel_group_id: None,
            updated_at: None,
        }],
        variables: std::collections::HashMap::new(),
        state: InstanceState::Running,
        version: 0,
    };
    repo.save(&inst).await.unwrap();

    let n = 16;
    let mut handles = Vec::with_capacity(n);
    for _ in 0..n {
        let r = Arc::clone(&repo);
        let iid = instance_id.clone();
        let tid = token_id.clone();
        handles.push(tokio::spawn(async move {
            r.claim_token(&iid, &tid, 0).await.unwrap_or(false)
        }));
    }
    let mut results = Vec::with_capacity(n);
    for h in handles {
        results.push(h.await.unwrap());
    }
    let success_count = results.iter().filter(|&&b| b).count();
    assert_eq!(
        success_count, 1,
        "single owner: exactly one claim should succeed per token"
    );
}