Code organization¶

Organize your code so that your workflows are composable and testable.

Separate orchestration from business logic¶

Business logic is the work a step performs. Orchestration is when and in what order steps run.

Put your logic in its own testable functions that the durable operation calls, rather than inline in the operation. Keep DurableContext out of your domain logic.

TypeScriptPythonJavaGoRustC#

// Business logic: no DurableContext, pure work.
async function validateOrder(order: Order): Promise<ValidationResult> {
  return orderValidator.validate(order);
}

async function chargePayment(order: Order): Promise<Receipt> {
  return paymentService.charge(order.total, order.cardToken);
}

async function scheduleShipment(order: Order): Promise<ShipmentId> {
  return shipmentService.schedule(order.id, order.address);
}

// Orchestration: the handler reads as a sequence of intent.
export const handler = withDurableExecution(
  async (order: Order, context: DurableContext) => {
    await context.step("validate", () => validateOrder(order));
    const receipt = await context.step("charge", () => chargePayment(order));
    const shipmentId = await context.step("schedule", () => scheduleShipment(order));
    return { receipt, shipmentId };
  },
);

def validate_order(order: dict) -> dict:
    return order_validator.validate(order)


def charge_payment(order: dict) -> dict:
    return payment_service.charge(order["total"], order["card_token"])


def schedule_shipment(order: dict) -> str:
    return shipment_service.schedule(order["id"], order["address"])


@durable_step
def validate_step(ctx: StepContext, order: dict) -> dict:
    return validate_order(order)


@durable_step
def charge_step(ctx: StepContext, order: dict) -> dict:
    return charge_payment(order)


@durable_step
def schedule_step(ctx: StepContext, order: dict) -> str:
    return schedule_shipment(order)


@durable_execution
def handler(order: dict, context: DurableContext) -> dict:
    context.step(validate_step(order))
    receipt = context.step(charge_step(order))
    shipment_id = context.step(schedule_step(order))
    return {"receipt": receipt, "shipmentId": shipment_id}

// Business logic lives on service classes, not the handler.
public class OrderWorkflow implements DurableHandler<Order, OrderResult> {
    private final OrderValidator validator;
    private final PaymentService payments;
    private final ShipmentService shipments;

    @Override
    public OrderResult handle(Order order, DurableContext context) {
        context.step("validate", ValidationResult.class,
            ctx -> validator.validate(order));

        Receipt receipt = context.step("charge", Receipt.class,
            ctx -> payments.charge(order.total(), order.cardToken()));

        String shipmentId = context.step("schedule", String.class,
            ctx -> shipments.schedule(order.id(), order.address()));

        return new OrderResult(receipt, shipmentId);
    }
}

Domain functions take plain arguments and return (T, error). Only the step body closure receives a durable.StepContext, so the domain logic stays free of SDK types.

package main

import "github.com/aws/aws-durable-execution-sdk-go/durable"

type Order struct {
    ID        string `json:"id"`
    Total     int    `json:"total"`
    CardToken string `json:"cardToken"`
    Address   string `json:"address"`
}

type ValidationResult struct {
    Valid bool `json:"valid"`
}

type Receipt struct {
    ID     string `json:"id"`
    Amount int    `json:"amount"`
}

type Result struct {
    Receipt    Receipt `json:"receipt"`
    ShipmentID string  `json:"shipmentId"`
}

// Business logic takes plain arguments and returns plain values. No durable context.
func validateOrder(order Order) (ValidationResult, error) {
    return ValidationResult{Valid: order.Total > 0}, nil
}

func chargePayment(order Order) (Receipt, error) {
    return Receipt{ID: "rcpt-" + order.ID, Amount: order.Total}, nil
}

func scheduleShipment(order Order) (string, error) {
    return "ship-" + order.ID, nil
}

// The handler orchestrates: it decides when each step runs.
func handler(ctx durable.Context, order Order) (Result, error) {
    if _, err := durable.Step(ctx, "validate", func(durable.StepContext) (ValidationResult, error) {
        return validateOrder(order)
    }); err != nil {
        return Result{}, err
    }
    receipt, err := durable.Step(ctx, "charge", func(durable.StepContext) (Receipt, error) {
        return chargePayment(order)
    })
    if err != nil {
        return Result{}, err
    }
    shipmentID, err := durable.Step(ctx, "schedule", func(durable.StepContext) (string, error) {
        return scheduleShipment(order)
    })
    if err != nil {
        return Result{}, err
    }
    return Result{Receipt: receipt, ShipmentID: shipmentID}, nil
}

func main() {
    durable.Start(handler)
}

Domain functions take plain arguments and return plain values. Only the step closure receives a durable::StepContext, so the domain logic stays free of SDK types. A step closure must be 'static, so each step moves in its own clone of order.

use aws_durable_execution_sdk as durable;
use serde::{Deserialize, Serialize};

#[derive(Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct Order {
    id: String,
    total: f64,
    card_token: String,
    address: String,
}

#[derive(Serialize, Deserialize)]
struct ValidationResult {
    valid: bool,
}

#[derive(Serialize, Deserialize)]
struct Receipt {
    id: String,
    amount: f64,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct OrderResult {
    receipt: Receipt,
    shipment_id: String,
}

// Business logic: no DurableContext, pure work.
fn validate_order(order: &Order) -> ValidationResult {
    order_validator_validate(order)
}

fn charge_payment(order: &Order) -> Receipt {
    payment_service_charge(order.total, &order.card_token)
}

fn schedule_shipment(order: &Order) -> String {
    shipment_service_schedule(&order.id, &order.address)
}

// Orchestration: the handler reads as a sequence of intent.
async fn handler(
    order: Order,
    ctx: durable::DurableContext,
) -> Result<OrderResult, durable::BoxError> {
    let to_validate = order.clone();
    ctx.step(move |_| async move { Ok(validate_order(&to_validate)) })
        .name("validate")
        .await?;
    let to_charge = order.clone();
    let receipt = ctx
        .step(move |_| async move { Ok(charge_payment(&to_charge)) })
        .name("charge")
        .await?;
    let shipment_id = ctx
        .step(move |_| async move { Ok(schedule_shipment(&order)) })
        .name("schedule")
        .await?;
    Ok(OrderResult {
        receipt,
        shipment_id,
    })
}

fn order_validator_validate(order: &Order) -> ValidationResult {
    ValidationResult {
        valid: order.total > 0.0,
    }
}

fn payment_service_charge(total: f64, card_token: &str) -> Receipt {
    Receipt {
        id: format!("rcpt-{card_token}"),
        amount: total,
    }
}

fn shipment_service_schedule(order_id: &str, _address: &str) -> String {
    format!("ship-{order_id}")
}

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

using Amazon.Lambda.Core;
using Amazon.Lambda.DurableExecution;

// Business logic: no IDurableContext, pure work.
public static class OrderLogic
{
    public static ValidationResult ValidateOrder(Order order) => new(order.Id, Valid: true);

    public static Receipt ChargePayment(Order order) => new(order.Id, order.Total);

    public static string ScheduleShipment(Order order) => $"ship-{order.Id}";
}

// Orchestration: the workflow reads as a sequence of intent.
public class SeparateLogicExample
{
    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<Order, OrderResult>(Workflow, input, context);

    private async Task<OrderResult> Workflow(Order order, IDurableContext ctx)
    {
        await ctx.StepAsync(
            async (_, _) => OrderLogic.ValidateOrder(order),
            name: "validate");

        Receipt receipt = await ctx.StepAsync(
            async (_, _) => OrderLogic.ChargePayment(order),
            name: "charge");

        string shipmentId = await ctx.StepAsync(
            async (_, _) => OrderLogic.ScheduleShipment(order),
            name: "schedule");

        return new OrderResult(receipt, shipmentId);
    }
}

public record Order(string Id, decimal Total, string Address, string CardToken);
public record ValidationResult(string OrderId, bool Valid);
public record Receipt(string OrderId, decimal Amount);
public record OrderResult(Receipt Receipt, string ShipmentId);

Tip

Simplify your unit testing by not referencing DurableContext in your domain logic functions. Cover orchestration separately with the durable function testing framework.

Group operations with child contexts¶

A child context is a named scope that groups operations in the execution history. Inside a child context you can call any durable operation, including nested child contexts.

Use a child context when a block of work is one logical unit that is composed of several operations.

TypeScriptPythonJavaGoRustC#

await context.runInChildContext("process-order", async (child) => {
  await child.step("validate", () => validate(order));
  const receipt = await child.step("charge", () => charge(order));
  await child.step("schedule", () => schedule(order, receipt));
  return "ok";
});

from aws_durable_execution_sdk_python import durable_with_child_context


@durable_with_child_context
def process_order(child: DurableContext, order: dict) -> str:
    child.step(validate_step(order))
    child.step(charge_step(order))
    child.step(schedule_step(order))
    return "ok"


context.run_in_child_context(process_order(order), name="process-order")

context.runInChildContext("process-order", String.class, child -> {
    child.step("validate", ValidationResult.class,
        ctx -> validator.validate(order));
    Receipt receipt = child.step("charge", Receipt.class,
        ctx -> payments.charge(order.total(), order.cardToken()));
    child.step("schedule", String.class,
        ctx -> shipments.schedule(order.id(), order.address()));
    return "ok";
});

The body receives its own child durable.Context. Call every durable operation on that child. An operation on the outer context from inside the body fails with durable.ErrWrongContext and records nothing. When you run durablelint, the SDK's static analysis tool, its durablechildctx rule reports this call.

package main

import "github.com/aws/aws-durable-execution-sdk-go/durable"

type Order struct {
    ID    string `json:"id"`
    Total int    `json:"total"`
}

type Receipt struct {
    ID     string `json:"id"`
    Amount int    `json:"amount"`
}

func validate(order Order) (bool, error) { return order.Total > 0, nil }
func charge(order Order) (Receipt, error) {
    return Receipt{ID: "rcpt-" + order.ID, Amount: order.Total}, nil
}
func schedule(order Order, r Receipt) (string, error) { return "ship-" + r.ID, nil }

func handler(ctx durable.Context, order Order) (string, error) {
    return durable.RunInChildContext(ctx, "process-order", func(child durable.Context) (string, error) {
        if _, err := durable.Step(child, "validate", func(durable.StepContext) (bool, error) {
            return validate(order)
        }); err != nil {
            return "", err
        }
        receipt, err := durable.Step(child, "charge", func(durable.StepContext) (Receipt, error) {
            return charge(order)
        })
        if err != nil {
            return "", err
        }
        if _, err := durable.Step(child, "schedule", func(durable.StepContext) (string, error) {
            return schedule(order, receipt)
        }); err != nil {
            return "", err
        }
        return "ok", nil
    })
}

func main() {
    durable.Start(handler)
}

The closure receives its own child durable::DurableContext. Call every durable operation on that child. An operation on the outer context from inside the closure fails the execution with a context usage error.

use aws_durable_execution_sdk as durable;
use serde::{Deserialize, Serialize};

#[derive(Clone, Deserialize)]
struct Order {
    id: String,
    total: f64,
}

#[derive(Clone, Serialize, Deserialize)]
struct Receipt {
    id: String,
    amount: f64,
}

fn validate(order: &Order) -> bool {
    order.total > 0.0
}

fn charge(order: &Order) -> Receipt {
    Receipt {
        id: format!("rcpt-{}", order.id),
        amount: order.total,
    }
}

fn schedule(_order: &Order, receipt: &Receipt) -> String {
    format!("ship-{}", receipt.id)
}

async fn handler(order: Order, ctx: durable::DurableContext) -> Result<String, durable::BoxError> {
    let result = ctx
        .run_in_child_context(move |child| async move {
            let to_validate = order.clone();
            child
                .step(move |_| async move { Ok(validate(&to_validate)) })
                .name("validate")
                .await?;
            let to_charge = order.clone();
            let receipt = child
                .step(move |_| async move { Ok(charge(&to_charge)) })
                .name("charge")
                .await?;
            child
                .step(move |_| async move { Ok(schedule(&order, &receipt)) })
                .name("schedule")
                .await?;
            Ok("ok".to_owned())
        })
        .name("process-order")
        .await?;
    Ok(result)
}

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

using Amazon.Lambda.Core;
using Amazon.Lambda.DurableExecution;

public class ChildContextExample
{
    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<Order, string>(Workflow, input, context);

    private async Task<string> Workflow(Order order, IDurableContext ctx)
    {
        // Group the order-processing operations into one child context.
        return await ctx.RunInChildContextAsync(
            async (child, ct) =>
            {
                await child.StepAsync(
                    async (_, _) => Validate(order),
                    name: "validate");
                Receipt receipt = await child.StepAsync(
                    async (_, _) => Charge(order),
                    name: "charge");
                await child.StepAsync(
                    async (_, _) => Schedule(order, receipt),
                    name: "schedule");
                return "ok";
            },
            name: "process-order");
    }

    private static ValidationResult Validate(Order order) => new(order.Id, Valid: true);
    private static Receipt Charge(Order order) => new(order.Id, order.Total);
    private static string Schedule(Order order, Receipt receipt) => $"ship-{order.Id}";
}

public record Order(string Id, decimal Total);
public record ValidationResult(string OrderId, bool Valid);
public record Receipt(string OrderId, decimal Amount);

When several steps share the same retry strategy, timeout, or serdes, define the configuration once and reuse it. Use the name of a configuration object to make its intent clear.

TypeScriptPythonJavaGoRustC#

import { retryPresets, StepSemantics } from "@aws/durable-execution-sdk-js";

const paymentStepConfig = {
  semantics: StepSemantics.AtMostOncePerRetry,
  retryStrategy: () => ({ shouldRetry: false }),
};

const idempotentStepConfig = {
  retryStrategy: retryPresets.default,
};

await context.step("charge", () => chargePayment(order), paymentStepConfig);
await context.step("refund", () => refundPayment(order), paymentStepConfig);
await context.step("fetch-user", () => userStore.get(id), idempotentStepConfig);

from aws_durable_execution_sdk_python.config import StepConfig, StepSemantics
from aws_durable_execution_sdk_python.retries import RetryPresets

PAYMENT_CONFIG = StepConfig(
    step_semantics=StepSemantics.AT_MOST_ONCE_PER_RETRY,
    retry_strategy=RetryPresets.none(),
)
IDEMPOTENT_CONFIG = StepConfig(retry_strategy=RetryPresets.default())

context.step(charge_step(order), config=PAYMENT_CONFIG)
context.step(refund_step(order), config=PAYMENT_CONFIG)
context.step(fetch_user(id), config=IDEMPOTENT_CONFIG)

private static final StepConfig PAYMENT_CONFIG = StepConfig.builder()
    .semantics(StepSemantics.AT_MOST_ONCE_PER_RETRY)
    .retryStrategy(RetryStrategies.Presets.NO_RETRY)
    .build();

private static final StepConfig IDEMPOTENT_CONFIG = StepConfig.builder()
    .retryStrategy(RetryStrategies.Presets.DEFAULT)
    .build();

context.step("charge", Receipt.class,
    ctx -> payments.charge(order), PAYMENT_CONFIG);
context.step("refund", Receipt.class,
    ctx -> payments.refund(order), PAYMENT_CONFIG);
context.step("fetch-user", User.class,
    ctx -> users.get(id), IDEMPOTENT_CONFIG);

Step takes functional options. Collect the options in a []durable.StepOption once and pass it to each step with opts....

package main

import "github.com/aws/aws-durable-execution-sdk-go/durable"

type Order struct {
    ID    string `json:"id"`
    Total int    `json:"total"`
}

type Receipt struct {
    ID     string `json:"id"`
    Amount int    `json:"amount"`
}

type User struct {
    ID   string `json:"id"`
    Name string `json:"name"`
}

func chargePayment(order Order) (Receipt, error) {
    return Receipt{ID: "rcpt-" + order.ID, Amount: order.Total}, nil
}
func refundPayment(order Order) (Receipt, error) {
    return Receipt{ID: "rfnd-" + order.ID, Amount: -order.Total}, nil
}
func fetchUser(id string) (User, error) { return User{ID: id, Name: "user-" + id}, nil }

func handler(ctx durable.Context, order Order) (string, error) {
    // A side-effecting call runs at most once and never retries.
    paymentOpts := []durable.StepOption{
        durable.WithSemantics(durable.AtMostOncePerRetry),
        durable.WithRetry(durable.NoRetry()),
    }
    // An idempotent read retries with the default exponential backoff.
    idempotentOpts := []durable.StepOption{
        durable.WithRetry(durable.ExponentialBackoff()),
    }

    if _, err := durable.Step(ctx, "charge", func(durable.StepContext) (Receipt, error) {
        return chargePayment(order)
    }, paymentOpts...); err != nil {
        return "", err
    }
    if _, err := durable.Step(ctx, "refund", func(durable.StepContext) (Receipt, error) {
        return refundPayment(order)
    }, paymentOpts...); err != nil {
        return "", err
    }
    user, err := durable.Step(ctx, "fetch-user", func(durable.StepContext) (User, error) {
        return fetchUser(order.ID)
    }, idempotentOpts...)
    if err != nil {
        return "", err
    }
    return user.Name, nil
}

func main() {
    durable.Start(handler)
}

Each step setting is a builder method. Collect the settings once in a value of your own, such as the StepConfig struct here, and apply its fields with .semantics(..) and .retry_strategy_config(..). durable::StepSemantics is Copy. durable::builders::RetryStrategyConfig is Clone, so clone it for each step that reuses it.

use aws_durable_execution_sdk as durable;
use durable::StepSemantics;
use durable::builders::RetryStrategyConfig;
use serde::{Deserialize, Serialize};

#[derive(Clone, Deserialize)]
struct Order {
    id: String,
    total: f64,
}

#[derive(Deserialize)]
struct OrderEvent {
    order: Order,
    id: String,
}

#[derive(Serialize, Deserialize)]
struct Receipt {
    id: String,
    amount: f64,
}

#[derive(Serialize, Deserialize)]
struct User {
    id: String,
    name: String,
}

struct StepConfig {
    semantics: StepSemantics,
    retry_strategy: RetryStrategyConfig,
}

fn charge_payment(order: &Order) -> Receipt {
    Receipt {
        id: format!("rcpt-{}", order.id),
        amount: order.total,
    }
}

fn refund_payment(order: &Order) -> Receipt {
    Receipt {
        id: format!("rfnd-{}", order.id),
        amount: -order.total,
    }
}

fn user_store_get(id: &str) -> User {
    User {
        id: id.to_owned(),
        name: format!("user-{id}"),
    }
}

async fn handler(event: OrderEvent, ctx: durable::DurableContext) -> Result<(), durable::BoxError> {
    let payment_step_config = StepConfig {
        semantics: StepSemantics::AtMostOncePerRetry,
        retry_strategy: RetryStrategyConfig::no_retry(),
    };

    let idempotent_step_config = StepConfig {
        semantics: StepSemantics::default(),
        retry_strategy: RetryStrategyConfig::default_preset(),
    };

    let OrderEvent { order, id } = event;
    let to_charge = order.clone();
    ctx.step(move |_| async move { Ok(charge_payment(&to_charge)) })
        .name("charge")
        .semantics(payment_step_config.semantics)
        .retry_strategy_config(payment_step_config.retry_strategy.clone())
        .await?;
    ctx.step(move |_| async move { Ok(refund_payment(&order)) })
        .name("refund")
        .semantics(payment_step_config.semantics)
        .retry_strategy_config(payment_step_config.retry_strategy)
        .await?;
    ctx.step(move |_| async move { Ok(user_store_get(&id)) })
        .name("fetch-user")
        .semantics(idempotent_step_config.semantics)
        .retry_strategy_config(idempotent_step_config.retry_strategy)
        .await?;
    Ok(())
}

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

using Amazon.Lambda.Core;
using Amazon.Lambda.DurableExecution;

public class GroupConfigExample
{
    // Define shared configuration once and reuse it. The name makes the intent clear.
    private static readonly StepConfig PaymentConfig = new()
    {
        Semantics = StepSemantics.AtMostOncePerRetry,
        RetryStrategy = RetryStrategy.None,
    };

    private static readonly StepConfig IdempotentConfig = new()
    {
        RetryStrategy = RetryStrategy.Default,
    };

    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<OrderEvent, User>(Workflow, input, context);

    private async Task<User> Workflow(OrderEvent input, IDurableContext ctx)
    {
        await ctx.StepAsync(
            async (_, _) => Charge(input.Order),
            name: "charge",
            config: PaymentConfig);

        await ctx.StepAsync(
            async (_, _) => Refund(input.Order),
            name: "refund",
            config: PaymentConfig);

        return await ctx.StepAsync(
            async (_, _) => GetUser(input.UserId),
            name: "fetch-user",
            config: IdempotentConfig);
    }

    private static Receipt Charge(Order order) => new(order.Id, order.Total);
    private static Receipt Refund(Order order) => new(order.Id, -order.Total);
    private static User GetUser(string id) => new(id, "Ada");
}

public record Order(string Id, decimal Total);
public record OrderEvent(Order Order, string UserId);
public record Receipt(string OrderId, decimal Amount);
public record User(string Id, string Name);

Run independent work concurrently¶

Use parallel for a fixed number of named branches and map to iterate over a variable-length list. Both checkpoint durable progress and survive Lambda timeouts or sandbox crashes, unlike language-specific constructs such as Promise.all, asyncio.gather, or CompletableFuture.

TypeScriptPythonJavaGoRustC#

// A small number of fixed branches.
const { fx, weather, quote } = await context.parallel("enrich", [
  async (ctx) => ctx.step("fx", () => fxRates.latest()),
  async (ctx) => ctx.step("weather", () => weatherApi.get()),
  async (ctx) => ctx.step("quote", () => quoteApi.get()),
]);

// A variable number of items.
const results = await context.map(items, async (ctx, item) =>
  ctx.step("process", () => process(item)),
  { maxConcurrency: 10 },
);

# A small number of fixed branches.
results = context.parallel(
    [
        lambda ctx: ctx.step(fx_rates_latest()),
        lambda ctx: ctx.step(weather_api_get()),
        lambda ctx: ctx.step(quote_api_get()),
    ],
    name="enrich",
)

# A variable number of items.
processed = context.map(items, process, name="process-items")

// A small number of fixed branches.
var enrich = context.parallel("enrich")
    .step("fx", FxRates.class, ctx -> fxRates.latest())
    .step("weather", Weather.class, ctx -> weatherApi.get())
    .step("quote", Quote.class, ctx -> quoteApi.get())
    .run();

// A variable number of items.
BatchResult<Result> processed = context.map(
    "process-items",
    items,
    Result.class,
    (ctx, item) -> process(item));

Use durable.Parallel for a fixed set of named branches and durable.Map for a variable-length list. The map function also receives the item index.

package main

import "github.com/aws/aws-durable-execution-sdk-go/durable"

type Event struct {
    Items []string `json:"items"`
}

type Result struct {
    FX        string   `json:"fx"`
    Weather   string   `json:"weather"`
    Quote     string   `json:"quote"`
    Processed []string `json:"processed"`
}

func latestFX() (string, error)           { return "1.10", nil }
func getWeather() (string, error)         { return "sunny", nil }
func getQuote() (string, error)           { return "42", nil }
func process(item string) (string, error) { return "processed-" + item, nil }

func handler(ctx durable.Context, event Event) (Result, error) {
    // A small, fixed set of branches.
    enrich, err := durable.Parallel(ctx, "enrich", []durable.Branch[string]{
        {Name: "fx", Func: func(c durable.Context) (string, error) {
            return durable.Step(c, "fx", func(durable.StepContext) (string, error) { return latestFX() })
        }},
        {Name: "weather", Func: func(c durable.Context) (string, error) {
            return durable.Step(c, "weather", func(durable.StepContext) (string, error) { return getWeather() })
        }},
        {Name: "quote", Func: func(c durable.Context) (string, error) {
            return durable.Step(c, "quote", func(durable.StepContext) (string, error) { return getQuote() })
        }},
    })
    if err != nil {
        return Result{}, err
    }

    // A variable-length list, run with a concurrency limit.
    processed, err := durable.Map(ctx, "process", event.Items,
        func(c durable.Context, item string, index int) (string, error) {
            return durable.Step(c, "process", func(durable.StepContext) (string, error) { return process(item) })
        },
        durable.WithMaxConcurrency(10),
    )
    if err != nil {
        return Result{}, err
    }

    fx, _ := enrich.Result("fx")
    weather, _ := enrich.Result("weather")
    quote, _ := enrich.Result("quote")
    return Result{FX: fx, Weather: weather, Quote: quote, Processed: processed.Results()}, nil
}

func main() {
    durable.Start(handler)
}

Use ctx.parallel for a fixed set of branches and ctx.map for a variable-length list. Every branch of one parallel returns the same type. The map closure also receives the item index. Awaiting either builder returns the results of the succeeded items as a Vec, in input order. Under the default completion config, any failed item makes the await return that item's error.

use aws_durable_execution_sdk as durable;
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct EnrichEvent {
    items: Vec<String>,
}

#[derive(Serialize)]
struct EnrichResult {
    fx: String,
    weather: String,
    quote: String,
    results: Vec<String>,
}

fn fx_rates_latest() -> String {
    "1.10".to_owned()
}

fn weather_api_get() -> String {
    "sunny".to_owned()
}

fn quote_api_get() -> String {
    "42".to_owned()
}

fn process(item: &str) -> String {
    format!("processed-{item}")
}

async fn handler(
    event: EnrichEvent,
    ctx: durable::DurableContext,
) -> Result<EnrichResult, durable::BoxError> {
    // A small number of fixed branches.
    let enriched = ctx
        .parallel([
            durable::Branch::unnamed(|ctx| async move {
                Ok(ctx
                    .step(|_| async { Ok(fx_rates_latest()) })
                    .name("fx")
                    .await?)
            }),
            durable::Branch::unnamed(|ctx| async move {
                Ok(ctx
                    .step(|_| async { Ok(weather_api_get()) })
                    .name("weather")
                    .await?)
            }),
            durable::Branch::unnamed(|ctx| async move {
                Ok(ctx
                    .step(|_| async { Ok(quote_api_get()) })
                    .name("quote")
                    .await?)
            }),
        ])
        .name("enrich")
        .await?;
    let Ok::<[String; 3], _>([fx, weather, quote]) = enriched.try_into() else {
        return Err("enrich returned an unexpected number of results".into());
    };

    // A variable number of items.
    let results = ctx
        .map(event.items, |ctx, item: String, _index| async move {
            Ok(ctx
                .step(move |_| async move { Ok(process(&item)) })
                .name("process")
                .await?)
        })
        .max_concurrency(10)
        .await?;

    Ok(EnrichResult {
        fx,
        weather,
        quote,
        results,
    })
}

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

using Amazon.Lambda.Core;
using Amazon.Lambda.DurableExecution;

public class ParallelismExample
{
    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<ItemsEvent, IReadOnlyList<string>>(
            Workflow, input, context);

    private async Task<IReadOnlyList<string>> Workflow(ItemsEvent input, IDurableContext ctx)
    {
        // A small number of fixed, named branches.
        IBatchResult<string> enrich = await ctx.ParallelAsync(
            new[]
            {
                new DurableBranch<string>("fx", async (branch, ct) =>
                    await branch.StepAsync(async (_, _) => FxRatesLatest(), name: "fx")),
                new DurableBranch<string>("weather", async (branch, ct) =>
                    await branch.StepAsync(async (_, _) => WeatherApiGet(), name: "weather")),
                new DurableBranch<string>("quote", async (branch, ct) =>
                    await branch.StepAsync(async (_, _) => QuoteApiGet(), name: "quote")),
            },
            name: "enrich");

        // A variable number of items.
        IBatchResult<string> processed = await ctx.MapAsync(
            input.Items,
            async (itemCtx, item, index, items, ct) =>
                await itemCtx.StepAsync(async (_, _) => Process(item), name: $"process-{index}"),
            name: "process-items",
            config: new MapConfig<string> { MaxConcurrency = 10 });

        return enrich.GetResults().Concat(processed.GetResults()).ToList();
    }

    private static string FxRatesLatest() => "fx";
    private static string WeatherApiGet() => "weather";
    private static string QuoteApiGet() => "quote";
    private static string Process(string item) => $"processed-{item}";
}

public record ItemsEvent(IReadOnlyList<string> Items);

See the parallel and map references for completion policies, concurrency limits, and per-item configuration.

Code organization¶

Separate orchestration from business logic¶

Group operations with child contexts¶

Group related configuration¶

Run independent work concurrently¶

See also¶