durable-rust

An idiomatic Rust SDK for AWS Lambda Durable Execution, providing full feature parity with the official AWS Python Durable Lambda SDK.

Why Rust for Durable Lambdas?

4-8x lower memory (~16-32MB vs Python's ~128MB baseline)
Order-of-magnitude faster cold starts (<100ms vs Python's typical 500ms+)
Direct cost savings at scale — billions of daily invocations means millions in annual compute reduction
Full behavioral compliance with the Python SDK — zero divergence in operation semantics

Features

All 8 core durable operations are supported:

Operation	Description
Step	Checkpointed work unit with optional retries
Wait	Time-based suspension
Callback	External signal coordination point
Invoke	Durable Lambda-to-Lambda invocation
Parallel	Concurrent fan-out with independent branches
Map	Parallel collection processing with batching
Child Context	Isolated subflow with its own checkpoint namespace
Logging	Replay-safe structured logging (no-op during replay)

Additionally:

Replay engine with deterministic operation ID generation (blake2b)
MockDurableContext for local testing without AWS credentials
4 API styles — choose the ergonomics that fit your team
Python-Rust compliance suite verifying identical behavior
Container deployment targeting provided.al2023

Quick Start

1. Add the dependency

Choose one API style crate (all are behaviorally identical):

[dependencies]
durable-lambda-closure = { path = "crates/durable-lambda-closure" }
tokio = { version = "1", features = ["full"] }
serde_json = "1"
lambda_runtime = "1.1"

2. Write a handler

use durable_lambda_closure::prelude::*;

async fn handler(
    event: serde_json::Value,
    mut ctx: ClosureContext,
) -> Result<serde_json::Value, DurableError> {
    // Step 1: Validate (checkpointed — replays from cache on re-invocation)
    let order: Result<serde_json::Value, String> = ctx.step("validate", || async {
        Ok(serde_json::json!({"order_id": 42, "valid": true}))
    }).await?;

    // Step 2: Charge payment
    let payment: Result<String, String> = ctx.step("charge", || async {
        Ok("tx-abc-123".to_string())
    }).await?;

    Ok(serde_json::json!({
        "order": order.unwrap(),
        "transaction": payment.unwrap(),
    }))
}

#[tokio::main]
async fn main() -> Result<(), lambda_runtime::Error> {
    durable_lambda_closure::run(handler).await
}

3. Write tests (no AWS credentials needed)

use durable_lambda_testing::prelude::*;

#[tokio::test]
async fn test_handler_replays_correctly() {
    let (mut ctx, calls, _ops) = MockDurableContext::new()
        .with_step_result("validate", r#"{"order_id": 42, "valid": true}"#)
        .with_step_result("charge", r#""tx-abc-123""#)
        .build()
        .await;

    // Steps replay cached results — closures are NOT executed
    let order: Result<serde_json::Value, String> = ctx
        .step("validate", || async { panic!("not executed during replay") })
        .await
        .unwrap();
    assert_eq!(order.unwrap()["order_id"], 42);

    // No checkpoints made during pure replay
    assert_no_checkpoints(&calls).await;
}

API Styles

All four approaches expose identical operations and produce identical behavior. Choose based on team preference:

Closure-Native (recommended default)

use durable_lambda_closure::prelude::*;

async fn handler(event: serde_json::Value, mut ctx: ClosureContext)
    -> Result<serde_json::Value, DurableError>
{
    let result: Result<i32, String> = ctx.step("work", || async { Ok(42) }).await?;
    Ok(serde_json::json!({"result": result.unwrap()}))
}

#[tokio::main]
async fn main() -> Result<(), lambda_runtime::Error> {
    durable_lambda_closure::run(handler).await
}

Proc-Macro

use durable_lambda_core::context::DurableContext;
use durable_lambda_core::error::DurableError;
use durable_lambda_macro::durable_execution;

#[durable_execution]
async fn handler(event: serde_json::Value, mut ctx: DurableContext)
    -> Result<serde_json::Value, DurableError>
{
    let result: Result<i32, String> = ctx.step("work", || async { Ok(42) }).await?;
    Ok(serde_json::json!({"result": result.unwrap()}))
}
// main() is generated by the macro

Trait-Based

use async_trait::async_trait;
use durable_lambda_trait::prelude::*;

struct MyHandler;

#[async_trait]
impl DurableHandler for MyHandler {
    async fn handle(&self, event: serde_json::Value, mut ctx: TraitContext)
        -> Result<serde_json::Value, DurableError>
    {
        let result: Result<i32, String> = ctx.step("work", || async { Ok(42) }).await?;
        Ok(serde_json::json!({"result": result.unwrap()}))
    }
}

#[tokio::main]
async fn main() -> Result<(), lambda_runtime::Error> {
    durable_lambda_trait::run(MyHandler).await
}

Builder-Pattern

use durable_lambda_builder::prelude::*;

#[tokio::main]
async fn main() -> Result<(), lambda_runtime::Error> {
    durable_lambda_builder::handler(
        |event: serde_json::Value, mut ctx: BuilderContext| async move {
            let result: Result<i32, String> = ctx.step("work", || async { Ok(42) }).await?;
            Ok(serde_json::json!({"result": result.unwrap()}))
        },
    )
    .run()
    .await
}

Error Handling

Step results have two Result layers:

Outer Result<_, DurableError>: SDK infrastructure errors (checkpoint failures, replay mismatches, AWS errors). Use ? to propagate these.
Inner Result<T, E>: Your business logic result. Both Ok and Err are checkpointed and replayed identically.

// The `?` on `.await?` propagates the outer DurableError (infrastructure layer).
// The returned value is your inner Result<T, E> — both arms are checkpointed.
let payment: Result<String, PaymentError> = ctx
    .step("charge", || async { charge_card().await })
    .await?;  // ? propagates DurableError (outer layer)

match payment {
    Ok(tx_id) => {
        // Step succeeded — tx_id is the checkpointed return value
    }
    Err(biz_err) => {
        // Step returned Err(PaymentError) — the error is also checkpointed
        // and will replay identically on re-invocation
    }
}

// Full three-arm pattern when you also handle infrastructure errors:
match ctx.step("charge", || async { charge_card().await }).await {
    Ok(Ok(tx_id))    => { /* business success */ }
    Ok(Err(biz_err)) => { /* business failure, checkpointed */ }
    Err(durable_err) => { /* SDK error: checkpoint fail, replay mismatch, etc. */ }
}

Operations Guide

Step (checkpointed work)

// Basic step
let result: Result<String, String> = ctx.step("validate", || async {
    Ok("valid".to_string())
}).await?;

// Step with retries
let result: Result<i32, String> = ctx.step_with_options(
    "charge",
    StepOptions::new().retries(3).backoff_seconds(5),
    || async { Ok(100) },
).await?;

Wait (time-based suspension)

ctx.wait("cooldown", 30).await?; // suspends for 30 seconds

Callback (external signal)

let handle = ctx.create_callback("approval", CallbackOptions::new()
    .timeout_seconds(300)
    .heartbeat_timeout_seconds(30)
).await?;

println!("Send approval to callback_id: {}", handle.callback_id);

let result: String = ctx.callback_result(&handle)?;

Invoke (Lambda-to-Lambda)

let result: serde_json::Value = ctx.invoke(
    "charge_payment",
    "payment-processor-lambda",
    &serde_json::json!({"order_id": 42, "amount": 99.99}),
).await?;

Parallel (concurrent fan-out)

use std::pin::Pin;
use std::future::Future;

// Why the type alias and Box::pin?
// `parallel()` requires a Vec of type-erased closures because each branch may have
// a different concrete future type (different captures, different return paths).
// `Box<dyn FnOnce(DurableContext) -> Pin<Box<dyn Future<...> + Send>>>` is the
// standard trait-object pattern for heterogeneous async closures.
// The BranchFn type alias keeps signatures readable.
type BranchFn = Box<dyn FnOnce(DurableContext)
    -> Pin<Box<dyn Future<Output = Result<i32, DurableError>> + Send>> + Send>;

let branches: Vec<BranchFn> = vec![
    Box::new(|mut ctx| Box::pin(async move {
        let r: Result<i32, String> = ctx.step("a", || async { Ok(10) }).await?;
        Ok(r.unwrap())
    })),
    Box::new(|mut ctx| Box::pin(async move {
        let r: Result<i32, String> = ctx.step("b", || async { Ok(20) }).await?;
        Ok(r.unwrap())
    })),
];

let result = ctx.parallel("fan_out", branches, ParallelOptions::new()).await?;
// result.results[0].result == Some(10)
// result.results[1].result == Some(20)

Map (parallel collection processing)

let items = vec![1, 2, 3, 4, 5];
let result = ctx.map(
    "process_items",
    items,
    MapOptions::new().batch_size(2), // process 2 at a time
    |item: i32, mut child_ctx: DurableContext| async move {
        let r: Result<i32, String> = child_ctx
            .step("double", || async move { Ok(item * 2) })
            .await?;
        Ok(r.unwrap())
    },
).await?;
// result.results: [2, 4, 6, 8, 10]

Child Context (isolated subflow)

let payment_result: serde_json::Value = ctx.child_context(
    "payment_flow",
    |mut child_ctx: DurableContext| async move {
        let r: Result<serde_json::Value, String> = child_ctx.step("charge", || async {
            Ok(serde_json::json!({"tx_id": "TX-42", "charged": true}))
        }).await?;
        Ok(r.unwrap())
    },
).await?;

Replay-Safe Logging

ctx.log("processing order");
ctx.log_with_data("order details", &serde_json::json!({"id": 42}));
ctx.log_debug("debug info");
ctx.log_warn("something unexpected");
ctx.log_error("operation failed");
// All log methods have _with_data variants for structured data

Advanced Features

These features extend the core operations with production-grade capabilities. Currently demonstrated in the closure-style examples.

Step Timeout

Enforce per-step deadlines. If the closure does not complete within the specified duration, the step fails with a timeout error.

let result: Result<String, String> = ctx.step_with_options(
    "external_call",
    StepOptions::new().timeout_seconds(10),
    || async { call_slow_service().await },
).await?;

Conditional Retry

Gate retries on a predicate. The retry budget is only consumed when the predicate returns true, allowing permanent failures to fail fast.

let result: Result<String, ApiError> = ctx.step_with_options(
    "api_call",
    StepOptions::new()
        .retries(3)
        .retry_if(|e: &ApiError| e.is_transient()),
    || async { call_api().await },
).await?;

Batch Checkpoint

Reduce checkpoint calls by up to 90% for sequences of independent steps. Instead of checkpointing after every step, multiple steps share a single batch checkpoint.

ctx.enable_batch_mode();

// These steps batch their checkpoints together
let a: Result<i32, String> = ctx.step("step_a", || async { Ok(1) }).await?;
let b: Result<i32, String> = ctx.step("step_b", || async { Ok(2) }).await?;
let c: Result<i32, String> = ctx.step("step_c", || async { Ok(3) }).await?;

Saga / Compensation

Register compensation (rollback) closures alongside forward operations. If a later step fails, call ctx.run_compensations() to execute all registered compensations in reverse order.

// Forward step with compensation
ctx.step_with_compensation(
    "charge_payment",
    || async { charge_card().await },
    || async { refund_card().await },
).await?;

ctx.step_with_compensation(
    "reserve_inventory",
    || async { reserve_items().await },
    || async { release_items().await },
).await?;

// On failure, roll back in reverse order
if should_rollback {
    ctx.run_compensations().await?;
}

Determinism Rules

Code outside durable operations re-executes on every invocation, including replays. Non-deterministic code produces different values each time, breaking replay.

Do / Don't

Non-deterministic source	Wrong (outside step)	Right (inside step)
Current time	`let now = Utc::now();`	`ctx.step("now", \|\| async { Ok(Utc::now()) }).await?`
Random values	`let id = Uuid::new_v4();`	`ctx.step("id", \|\| async { Ok(Uuid::new_v4()) }).await?`
Random numbers	`let n = rand::random::<u32>();`	`ctx.step("rng", \|\| async { Ok(rand::random::<u32>()) }).await?`

Wrong:

// BAD: Uuid changes on every invocation — replay gets a different ID
let order_id = Uuid::new_v4();
let result: Result<(), String> = ctx.step("create_order", || async move {
    create_order(order_id).await // Different ID on replay!
}).await?;

Right:

// GOOD: Uuid is generated inside the step — same value replayed every time
let order_id_result: Result<Uuid, String> = ctx.step("gen_id", || async {
    Ok(Uuid::new_v4())
}).await?;
let order_id = order_id_result.unwrap();

let result: Result<(), String> = ctx.step("create_order", || async move {
    create_order(order_id).await // Same value on replay
}).await?;

Safety checklist:

No Utc::now() / SystemTime::now() outside a step
No Uuid::new_v4() outside a step
No rand::random() or rand::thread_rng() outside a step
No environment variable reads that may differ between invocations outside a step
Operation order is fixed — do not reorder operations between deployments of in-flight workflows

Testing

The durable-lambda-testing crate provides MockDurableContext for writing tests without AWS credentials.

Mock Builder Methods

MockDurableContext::new()
    .with_step_result("name", r#"json_result"#)     // successful step
    .with_step_error("name", "ErrorType", r#"json"#) // failed step
    .with_wait("name")                                // completed wait
    .with_callback("name", "cb-id", r#"json"#)       // signaled callback
    .with_invoke("name", r#"json_result"#)            // completed invoke
    .build()
    .await
// Returns: (DurableContext, CheckpointRecorder, OperationRecorder)

Assertion Helpers

assert_no_checkpoints(&calls).await;           // pure replay verification
assert_checkpoint_count(&calls, 2).await;      // exact checkpoint count
assert_operations(&ops, &["step:validate", "step:charge"]).await; // operation sequence
assert_operation_names(&ops, &["validate", "charge"]).await;      // names only
assert_operation_count(&ops, 3).await;          // total operations

Running Tests

# Run all tests
cargo test --workspace

# Run specific test crate
cargo test -p e2e-tests           # end-to-end workflows
cargo test -p parity-tests        # cross-approach parity
cargo test -p durable-lambda-compliance  # Python-Rust compliance

# Run tests for a specific crate
cargo test -p durable-lambda-core
cargo test -p durable-lambda-testing

Project Structure

durable-rust/
├── crates/
│   ├── durable-lambda-core/       # Replay engine, types, errors, all operations
│   ├── durable-lambda-macro/      # #[durable_execution] proc-macro
│   ├── durable-lambda-closure/    # Closure-native API (ClosureContext)
│   ├── durable-lambda-trait/      # Trait-based API (TraitContext + DurableHandler)
│   ├── durable-lambda-builder/    # Builder-pattern API (BuilderContext)
│   └── durable-lambda-testing/    # MockDurableContext, assertions
├── examples/
│   ├── closure-style/             # 15 examples: all operations + advanced features ([README](examples/closure-style/README.md))
│   ├── macro-style/               # 11 examples: all core operations ([README](examples/macro-style/README.md))
│   ├── trait-style/               # 11 examples: all core operations ([README](examples/trait-style/README.md))
│   └── builder-style/             # 11 examples: all core operations ([README](examples/builder-style/README.md))
├── tests/
│   ├── e2e/                       # 28 end-to-end workflow tests
│   └── parity/                    # Cross-approach behavioral parity tests
├── compliance/
│   ├── python/                    # Python reference workflows
│   ├── rust/                      # Rust equivalent workflows
│   └── tests/fixtures/            # Shared operation sequence fixtures
└── docs/
    └── migration-guide.md         # Python-to-Rust migration guide

Crate Dependency Graph

durable-lambda-closure ─┐
durable-lambda-macro  ──┤
durable-lambda-trait  ──┼── durable-lambda-core
durable-lambda-builder ─┤
durable-lambda-testing ─┘

All approach crates depend only on durable-lambda-core. No circular or cross-approach dependencies.

Python Migration

See docs/migration-guide.md for a detailed guide covering:

Conceptual mapping table (Python operation → Rust equivalent)
Handler registration patterns
Type system differences
Error handling
Testing migration
Container deployment

Key mapping:

Python	Rust
`context.call_activity("name", fn)`	`ctx.step("name", \|\| async { ... }).await?`
`context.create_wait("name", secs)`	`ctx.wait("name", secs).await?`
`context.parallel("name", branches)`	`ctx.parallel("name", branches, opts).await?`
`MockContext()`	`MockDurableContext::new().build().await`

Troubleshooting

Send + 'static on parallel/map closures

Problem: Closures in parallel() or map() capture a borrowed reference (&T), violating the Send + 'static requirement imposed by tokio::spawn.

Compiler error (representative):

error[E0521]: borrowed data escapes outside of closure
  --> src/main.rs:15:9
   |
   |     Box::new(|mut ctx| Box::pin(async move {
   |              --------- `data` is a reference that is only valid in the closure body
   |         process(data); // captured &data violates 'static
   |         ^^^^^^^^^^^^ `data` escapes the closure body here

Fix: Clone the data before the closure and use move:

let data = data.clone(); // owned copy
Box::new(move |mut ctx| Box::pin(async move {
    process(&data); // owned — satisfies Send + 'static
    Ok(())
}))

Serialize + DeserializeOwned bounds

Problem: A type flowing through step(), parallel(), map(), or child_context() does not derive Serialize + Deserialize. Both T and E in Result<T, E> must implement these traits.

Compiler error (representative):

error[E0277]: the trait bound `MyType: Serialize` is not satisfied
  --> src/main.rs:10:5
   |
   |     let result: Result<MyType, String> = ctx.step("work", || async {
   |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `Serialize` is not implemented for `MyType`

Fix: Add serde derives to every type used in durable operations:

#[derive(serde::Serialize, serde::Deserialize)]
struct MyType { /* ... */ }

Missing type annotations on step results

Problem: The compiler cannot infer T and E for a step result because the type information comes from serde deserialization, not from the closure return type alone.

Compiler error (representative):

error[E0284]: type annotations needed
  --> src/main.rs:10:9
   |
   |     let result = ctx.step("work", || async { Ok(42) }).await?;
   |         ^^^^^^ cannot infer type for type parameter `T` declared on the method `step`

Fix: Always annotate step results explicitly:

let result: Result<i32, String> = ctx.step("work", || async { Ok(42) }).await?;
//         ^^^^^^^^^^^^^^^^^^^ required — compiler cannot infer T and E

Container Deployment

Durable Lambdas deploy as container images using provided.al2023:

FROM rust:1-slim AS builder
WORKDIR /app
COPY . .
RUN cargo build --release

FROM public.ecr.aws/lambda/provided:al2023
COPY --from=builder /app/target/release/my-handler ${LAMBDA_RUNTIME_DIR}/bootstrap
CMD ["bootstrap"]

Requirements

Rust: Latest stable toolchain (1.82.0+)
Runtime: tokio (required by aws-sdk-lambda and lambda_runtime)
AWS SDK: aws-sdk-lambda 1.118+
For deployment: AWS account with durable execution enabled

For local development and testing, no AWS credentials are needed.

Contributing

This project follows 38 implementation rules documented in project-context.md. Read these before making changes.

License

Licensed under either of MIT or Apache-2.0 at your option.

Name		Name	Last commit message	Last commit date
.claude/skills		.claude/skills
.github/workflows		.github/workflows
.planning		.planning
_bmad-output		_bmad-output
_bmad		_bmad
compliance		compliance
crates		crates
docs		docs
examples		examples
infra		infra
scripts		scripts
tests		tests
.dockerignore		.dockerignore
.gitignore		.gitignore
CLAUDE.md		CLAUDE.md
Cargo.lock		Cargo.lock
Cargo.toml		Cargo.toml
LICENSE-APACHE		LICENSE-APACHE
LICENSE-MIT		LICENSE-MIT
README.md		README.md

Latest commit

History

Folders and files

Repository files navigation

durable-rust

Why Rust for Durable Lambdas?

Features

Quick Start

1. Add the dependency

2. Write a handler

3. Write tests (no AWS credentials needed)

API Styles

Closure-Native (recommended default)

Proc-Macro

Trait-Based

Builder-Pattern

Error Handling

Operations Guide

Step (checkpointed work)

Wait (time-based suspension)

Callback (external signal)

Invoke (Lambda-to-Lambda)

Parallel (concurrent fan-out)

Map (parallel collection processing)

Child Context (isolated subflow)

Replay-Safe Logging

Advanced Features

Step Timeout

Conditional Retry

Batch Checkpoint

Saga / Compensation

Determinism Rules

Do / Don't

Testing

Mock Builder Methods

Assertion Helpers

Running Tests

Project Structure

Crate Dependency Graph

Python Migration

Troubleshooting

Send + 'static on parallel/map closures

Serialize + DeserializeOwned bounds

Missing type annotations on step results

Container Deployment

Requirements

Contributing

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages