duroxide 0.1.27

Durable code execution framework for Rust
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
124
125
126
127
128
129
130

//! Tests for RuntimeOptions configuration
#![allow(clippy::unwrap_used)]
#![allow(clippy::clone_on_ref_ptr)]
#![allow(clippy::expect_used)]

use duroxide::runtime::registry::ActivityRegistry;
use duroxide::runtime::{self, RuntimeOptions};
use duroxide::{ActivityContext, Client, OrchestrationContext, OrchestrationRegistry};
use std::time::{Duration, Instant};

mod common;

/// Test: Runtime uses default polling frequency (10ms)
#[tokio::test]
async fn test_default_polling_frequency() {
    let (store, _temp_dir) = common::create_sqlite_store_disk().await;

    let activities = ActivityRegistry::builder()
        .register("QuickTask", |_ctx: ActivityContext, _: String| async move {
            Ok("done".to_string())
        })
        .build();

    let orch = |ctx: OrchestrationContext, _: String| async move { ctx.schedule_activity("QuickTask", "").await };

    let orchestrations = OrchestrationRegistry::builder().register("TestOrch", orch).build();

    // Start with default options (10ms polling)
    let rt = runtime::Runtime::start_with_store(store.clone(), activities, orchestrations).await;

    let client = Client::new(store.clone());
    let start_time = Instant::now();

    client
        .start_orchestration("test-default", "TestOrch", "")
        .await
        .unwrap();

    let status = client
        .wait_for_orchestration("test-default", std::time::Duration::from_secs(2))
        .await
        .unwrap();

    let elapsed = start_time.elapsed();

    assert!(matches!(status, runtime::OrchestrationStatus::Completed { .. }));
    // With 10ms polling, should complete reasonably fast (< 500ms for simple workflow)
    assert!(elapsed.as_millis() < 500, "Took too long: {}ms", elapsed.as_millis());

    rt.shutdown(None).await;
}

/// Test: Runtime uses custom polling frequency (50ms)
#[tokio::test]
async fn test_custom_polling_frequency() {
    let (store, _temp_dir) = common::create_sqlite_store_disk().await;

    let activities = ActivityRegistry::builder()
        .register("QuickTask", |_ctx: ActivityContext, _: String| async move {
            Ok("done".to_string())
        })
        .build();

    let orch = |ctx: OrchestrationContext, _: String| async move { ctx.schedule_activity("QuickTask", "").await };

    let orchestrations = OrchestrationRegistry::builder().register("TestOrch", orch).build();

    // Start with slower polling (50ms)
    let options = RuntimeOptions {
        dispatcher_min_poll_interval: Duration::from_millis(50),
        ..Default::default()
    };

    let rt = runtime::Runtime::start_with_options(store.clone(), activities, orchestrations, options).await;

    let client = Client::new(store.clone());

    client.start_orchestration("test-custom", "TestOrch", "").await.unwrap();

    let status = client
        .wait_for_orchestration("test-custom", std::time::Duration::from_secs(2))
        .await
        .unwrap();

    assert!(matches!(status, runtime::OrchestrationStatus::Completed { .. }));

    rt.shutdown(None).await;
}

/// Test: Fast polling (1ms) for high-throughput scenarios
#[tokio::test]
async fn test_fast_polling() {
    let (store, _temp_dir) = common::create_sqlite_store_disk().await;

    let activities = ActivityRegistry::builder()
        .register("Task", |_ctx: ActivityContext, _: String| async move {
            Ok("done".to_string())
        })
        .build();

    let orch = |ctx: OrchestrationContext, _: String| async move { ctx.schedule_activity("Task", "").await };

    let orchestrations = OrchestrationRegistry::builder().register("FastOrch", orch).build();

    // Very responsive: 1ms polling
    let options = RuntimeOptions {
        dispatcher_min_poll_interval: Duration::from_millis(1),
        ..Default::default()
    };

    let rt = runtime::Runtime::start_with_options(store.clone(), activities, orchestrations, options).await;

    let client = Client::new(store.clone());
    let start_time = Instant::now();

    client.start_orchestration("test-fast", "FastOrch", "").await.unwrap();

    let status = client
        .wait_for_orchestration("test-fast", std::time::Duration::from_secs(2))
        .await
        .unwrap();

    let elapsed = start_time.elapsed();

    assert!(matches!(status, runtime::OrchestrationStatus::Completed { .. }));
    // Fast polling should complete very quickly
    assert!(elapsed.as_millis() < 300, "Took too long: {}ms", elapsed.as_millis());

    rt.shutdown(None).await;
}