fusee 1.0.0

Embeddable durable execution at the speed of the disk: workflows whose steps are durable within microseconds of fdatasync
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

//! A payment saga: reserve stock, charge, ship; compensate on failure or
//! cancellation. Every external call gets `ctx.key()` as its idempotency key.
use fusee::{Config, Ctx, Engine, Error, RetryPolicy, StepOptions};
use serde::{Deserialize, Serialize};
use std::time::Duration;

#[derive(Serialize, Deserialize, Clone)]
struct Order {
    id: String,
    sku: String,
    cents: u64,
}

/// A short ID made from a step's idempotency key (as a service would): FNV-1a.
fn short(key: &str) -> String {
    let h = key.bytes().fold(0x811c_9dc5_u32, |h, b| {
        (h ^ u32::from(b)).wrapping_mul(0x0100_0193)
    });
    format!("{h:08x}")
}

// Stand-ins for real services. Each accepts an idempotency key.
async fn reserve(sku: &str, key: &str) -> Result<String, String> {
    Ok(format!("res-{sku}-{}", short(key)))
}
async fn charge(cents: u64, key: &str) -> Result<String, String> {
    if cents > 1_000_000 {
        return Err("card declined".into());
    }
    Ok(format!("ch-{}", short(key)))
}
async fn refund(charge: &str) -> Result<(), String> {
    println!("  refunded {charge}");
    Ok(())
}
async fn release(reservation: &str) -> Result<(), String> {
    println!("  released {reservation}");
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let engine = Engine::open(Config::local("./data/payment-saga")).await?;
    let saga = engine.register("checkout", |ctx: Ctx, order: Order| async move {
        let reservation: String = ctx
            .step("reserve", || {
                let (sku, key) = (order.sku.clone(), ctx.key());
                async move { reserve(&sku, &key).await }
            })
            .await?;
        let retry = StepOptions::default().retry(RetryPolicy::exponential(
            5,
            Duration::from_millis(200),
            Duration::from_secs(60),
        ));
        let charged: Result<String, Error> = ctx
            .step("charge", || {
                let key = ctx.key();
                async move { charge(order.cents, &key).await }
            })
            .options(retry)
            .await;
        let charge_id = match charged {
            Ok(c) => c,
            Err(e) => {
                ctx.step("release", || {
                    let r = reservation.clone();
                    async move { release(&r).await }
                })
                .await?;
                return Err(e);
            }
        };
        // Wait for the warehouse to confirm, at most a day.
        match ctx
            .wait_for::<bool>("shipped")
            .timeout(Duration::from_secs(86_400))
            .await
        {
            Ok(Some(true)) => Ok(charge_id),
            Ok(_) | Err(Error::Cancelled(_)) => {
                ctx.step("refund", || {
                    let c = charge_id.clone();
                    async move { refund(&c).await }
                })
                .await?;
                ctx.step("release", || {
                    let r = reservation.clone();
                    async move { release(&r).await }
                })
                .await?;
                Err(Error::Cancelled("not shipped; refunded".into()))
            }
            Err(e) => Err(e),
        }
    });
    let order = Order {
        id: "order-123".into(),
        sku: "SKU-9".into(),
        cents: 4_200,
    };
    let run = engine.start(&saga, &order.id, &order).await?;
    run.signal("shipped", &true).await?;
    println!("charged: {}", run.result().await?);
    engine.shutdown().await
}