a3s-flow 0.10.15

Durable workflow engine and Rust SDK for A3S
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

#[cfg(feature = "postgres")]
use a3s_flow::PostgresFlowTaskQueue;
use a3s_flow::{FlowTask, FlowTaskQueue, LocalFileFlowTaskQueue};
use chrono::{DateTime, Utc};
use std::sync::Arc;
#[cfg(feature = "postgres")]
use uuid::Uuid;

fn task(run_id: &str) -> FlowTask {
    FlowTask::DriveRun {
        run_id: run_id.to_string(),
    }
}

#[tokio::test]
async fn local_queue_minimum_cutoff_does_not_panic_or_requeue_a_lease() {
    let directory = tempfile::tempdir().unwrap();
    let queue = Arc::new(LocalFileFlowTaskQueue::new(directory.path()));
    queue.enqueue(task("minimum-cutoff")).await.unwrap();
    let lease = queue.lease().await.unwrap().unwrap();

    let reclaimer = queue.clone();
    let requeued = tokio::spawn(async move {
        reclaimer
            .requeue_inflight_older_than(DateTime::<Utc>::MIN_UTC)
            .await
    })
    .await
    .expect("minimum cutoff handling must not panic")
    .unwrap();

    assert_eq!(requeued, 0);
    assert_eq!(queue.inflight_len().await.unwrap(), 1);
    queue.ack(&lease.lease_id).await.unwrap();
}

#[tokio::test]
async fn local_queue_maximum_cutoff_does_not_panic_and_requeues_every_lease() {
    let directory = tempfile::tempdir().unwrap();
    let queue = Arc::new(LocalFileFlowTaskQueue::new(directory.path()));
    queue.enqueue(task("maximum-cutoff")).await.unwrap();
    queue.lease().await.unwrap().unwrap();

    let reclaimer = queue.clone();
    let requeued = tokio::spawn(async move {
        reclaimer
            .requeue_inflight_older_than(DateTime::<Utc>::MAX_UTC)
            .await
    })
    .await
    .expect("maximum cutoff handling must not panic")
    .unwrap();

    assert_eq!(requeued, 1);
    assert_eq!(queue.inflight_len().await.unwrap(), 0);
    assert_eq!(queue.dequeue().await.unwrap(), Some(task("maximum-cutoff")));
}

#[cfg(feature = "postgres")]
fn postgres_url_from_env() -> Option<String> {
    std::env::var("A3S_FLOW_POSTGRES_URL")
        .ok()
        .filter(|url| !url.trim().is_empty())
}

#[cfg(feature = "postgres")]
#[tokio::test]
async fn postgres_queue_extreme_cutoffs_preserve_ordering_when_url_is_configured() {
    let Some(postgres_url) = postgres_url_from_env() else {
        return;
    };
    let queue_name = format!("time-bounds-{}", Uuid::new_v4());
    let queue = PostgresFlowTaskQueue::connect_with_queue(&postgres_url, queue_name)
        .await
        .unwrap();
    queue.enqueue(task("postgres-cutoffs")).await.unwrap();
    queue.lease().await.unwrap().unwrap();

    assert_eq!(
        queue
            .requeue_inflight_older_than(DateTime::<Utc>::MIN_UTC)
            .await
            .unwrap(),
        0
    );
    assert_eq!(queue.inflight_len().await.unwrap(), 1);
    assert_eq!(
        queue
            .requeue_inflight_older_than(DateTime::<Utc>::MAX_UTC)
            .await
            .unwrap(),
        1
    );
    assert_eq!(
        queue.dequeue().await.unwrap(),
        Some(task("postgres-cutoffs"))
    );
}