Map¶

Apply a function to each item in a collection¶

Map executes a function for each item in a collection concurrently. It manages concurrency, collects results as items complete, and checkpoints the outcome.

Each item runs in its own child context. The default nested mode checkpoints that context and its result. Flat mode omits the per-item context checkpoint to reduce operation overhead.

Use map to apply the same operation to every item in a collection. Use parallel instead to execute different operations concurrently.

TypeScriptPythonJavaGoRustC#

import {
  BatchResult,
  DurableContext,
  withDurableExecution,
} from "@aws/durable-execution-sdk-js";

export const handler = withDurableExecution(
  async (event: any, context: DurableContext): Promise<number[]> => {
    const result: BatchResult<number> = await context.map(
      "square-numbers",
      [1, 2, 3, 4, 5],
      async (ctx, item, index) =>
        ctx.step(`square-${index}`, async () => item * item),
    );

    return result.getResults();
  },
);

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)


def square(ctx: DurableContext, item: int, index: int, items: list[int]) -> int:
    return ctx.step(lambda _: item * item, name=f"square-{index}")


@durable_execution
def handler(event: dict, context: DurableContext) -> list[int]:
    result: BatchResult[int] = context.map(
        [1, 2, 3, 4, 5],
        square,
        name="square-numbers",
    )
    return result.get_results()

import java.util.List;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.model.MapResult;

public class SimpleMap extends DurableHandler<Void, List<Integer>> {
    @Override
    public List<Integer> handleRequest(Void input, DurableContext context) {
        MapResult<Integer> result = context.map(
                "square-numbers",
                List.of(1, 2, 3, 4, 5),
                Integer.class,
                (item, index, ctx) -> ctx.step(
                        "square-" + index, Integer.class, s -> item * item));

        return result.results();
    }
}

// Package main squares each number in a collection with Map and returns the
// results in input order.
package main

import (
    "fmt"

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

func handler(ctx durable.Context, _ any) ([]int, error) {
    result, err := durable.Map(ctx, "square-numbers", []int{1, 2, 3, 4, 5},
        func(ctx durable.Context, item int, index int) (int, error) {
            return durable.Step(ctx, fmt.Sprintf("square-%d", index),
                func(durable.StepContext) (int, error) {
                    return item * item, nil
                })
        })
    if err != nil {
        return nil, err
    }
    return result.Results(), nil
}

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

use aws_durable_execution_sdk as durable;

async fn handler(
    _event: serde_json::Value,
    ctx: durable::DurableContext,
) -> Result<Vec<i64>, durable::BoxError> {
    let results = ctx
        .map(vec![1_i64, 2, 3, 4, 5], |ctx, item, index| async move {
            Ok(ctx
                .step(move |_| async move { Ok(item * item) })
                .name(format!("square-{index}"))
                .await?)
        })
        .name("square-numbers")
        .await?;

    Ok(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 SimpleMapExample
{
    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<object, IReadOnlyList<int>>(Workflow, input, context);

    private async Task<IReadOnlyList<int>> Workflow(object input, IDurableContext ctx)
    {
        IBatchResult<int> result = await ctx.MapAsync(
            new[] { 1, 2, 3, 4, 5 },
            async (itemCtx, item, index, items, ct) =>
                await itemCtx.StepAsync(
                    async (_, _) => item * item,
                    name: $"square-{index}"),
            name: "square-numbers");

        return result.GetResults();
    }
}

Method signature¶

context.map¶

TypeScriptPythonJavaGoRustC#

// Named overload
map<TInput, TOutput>(
  name: string | undefined,
  items: TInput[],
  mapFunc: MapFunc<TInput, TOutput>,
  config?: MapConfig<TInput, TOutput>,
): DurablePromise<BatchResult<TOutput>>

// Unnamed overload
map<TInput, TOutput>(
  items: TInput[],
  mapFunc: MapFunc<TInput, TOutput>,
  config?: MapConfig<TInput, TOutput>,
): DurablePromise<BatchResult<TOutput>>

Parameters:

name (optional) A name for the map operation. Pass undefined to omit.
items An array of items to process.
mapFunc A MapFunc called for each item. See Map Function.
config (optional) A MapConfig<TInput, TOutput> object.

Returns: DurablePromise<BatchResult<TOutput>>. Use await to get the result.

Throws: Item exceptions are captured in the BatchResult. Call throwIfError() to re-throw the first failure.

def map(
    inputs: Sequence[U],
    func: Callable[[DurableContext, U, int, Sequence[U]], T],
    name: str | None = None,
    config: MapConfig | None = None,
) -> BatchResult[T]: ...

Parameters:

inputs A sequence of items to process.
func A callable called for each item. See Map Function.
name (optional) A name for the map operation.
config (optional) A MapConfig object.

Returns: BatchResult[T].

Raises: Item exceptions are captured in the BatchResult. Call throw_if_error() to re-raise the first failure.

// sync — blocks until all items complete
<I, O> MapResult<O> map(String name, Collection<I> items, Class<O> resultType,
                        MapFunction<I, O> function)
<I, O> MapResult<O> map(String name, Collection<I> items, Class<O> resultType,
                        MapFunction<I, O> function, MapConfig config)
<I, O> MapResult<O> map(String name, Collection<I> items, TypeToken<O> resultType,
                        MapFunction<I, O> function)
<I, O> MapResult<O> map(String name, Collection<I> items, TypeToken<O> resultType,
                        MapFunction<I, O> function, MapConfig config)

// async — returns immediately
<I, O> DurableFuture<MapResult<O>> mapAsync(String name, Collection<I> items,
                                            Class<O> resultType, MapFunction<I, O> function)
<I, O> DurableFuture<MapResult<O>> mapAsync(String name, Collection<I> items,
                                            Class<O> resultType, MapFunction<I, O> function,
                                            MapConfig config)
<I, O> DurableFuture<MapResult<O>> mapAsync(String name, Collection<I> items,
                                            TypeToken<O> resultType, MapFunction<I, O> function)
<I, O> DurableFuture<MapResult<O>> mapAsync(String name, Collection<I> items,
                                            TypeToken<O> resultType, MapFunction<I, O> function,
                                            MapConfig config)

Parameters:

name (required) A name for the map operation.
items A Collection<I> of items to process. Its iteration order must remain stable during replay so item indexes match their checkpoints. Use an ordered collection such as List, LinkedHashSet, or TreeSet. The SDK rejects HashSet and the keySet(), values(), and entrySet() views of known unordered maps. Copy or sort unordered inputs into a List before calling map() or mapAsync().
resultType Class<O> or TypeToken<O> for deserialization.
function A MapFunction<I, O> called for each item. See Map Function.
config (optional) A MapConfig object.

Returns: MapResult<O> from map(), or DurableFuture<MapResult<O>> from mapAsync().

Throws: IllegalArgumentException if items is null or has a known non-deterministic iteration order. Item exceptions are captured in MapResult. Inspect failed() to detect failures. If the SDK cannot reconstruct the original exception, it throws MapIterationFailedException.

func Map[I, O any](ctx Context, name string, items []I, fn func(ctx Context, item I, index int) (O, error), opts ...BatchOption) (BatchResult[O], error)

Parameters:

ctx The durable context.
name (required) A name for the map operation. Pass "" to omit it.
items The []I slice of items to process.
fn The item function called for each item. See Map Function.
opts (optional) Zero or more BatchOption values. See MapConfig.

Returns: (BatchResult[O], error). Item failures are captured in the BatchResult.

Errors: The batch fails as a unit when its Reason is CompletionFailureToleranceExceeded or CompletionCustomFailed. Map then returns the populated BatchResult together with a *BatchError. It returns a *BatchCompletionError instead when a custom decision failed the batch and no item failed. A failure within the configured tolerance returns a nil error. Any other error (an invalid option, suspension, a checkpoint failure) comes with a zero BatchResult. Return it unchanged. See Error handling.

pub fn map<Items, I, O, F, Fut>(&self, items: Items, f: F) -> MapBuilder<I, O, F, Fut>
where
    Items: IntoIterator<Item = I>,
    F: Fn(DurableContext, I, usize) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = Result<O, BoxError>> + Send + 'static,
    I: Send + 'static,
    O: Send + 'static,
...

impl<I, O, F, Fut, IS, RS> IntoFuture for MapBuilder<I, O, F, Fut, IS, RS>
...
    type Output = Result<Vec<O>, OperationError>;
...

pub async fn await_batch(self) -> Result<BatchResult<O>, OperationError>
...
pub fn spawn(self) -> DurableFuture<Vec<O>>
...
pub fn spawn_batch(self) -> DurableFuture<BatchResult<O>>
...
pub fn future(self) -> DurableFuture<Vec<O>>
...

Parameters:

items Any IntoIterator of items to process. The SDK collects the items into a Vec when map is called. Its iteration order must be the same on every invocation so item indexes match their checkpoints. Sort values from a HashMap or HashSet into a Vec before calling map.
f The item function, called for each item. See Map Function.

map returns a MapBuilder. Set the name and the options with its methods, such as .name("process-items"), and then await it. Without .name(...) the map operation is unnamed. See MapConfig.

Returns: .await gives Result<Vec<O>, durable::OperationError>, the results of the succeeded items in input order. .await_batch().await gives the whole BatchResult<O>. See Result types. The map starts when the future is first polled. .future() returns a lazy durable::DurableFuture<Vec<O>> for ctx.join_all, ctx.try_join_all, ctx.race, or ctx.select_ok. .spawn() and .spawn_batch() start the map at once and return a durable::DurableFuture to await later.

Errors: .await returns an error when the batch completes with CompletionReason::FailureToleranceExceeded, which the default fail-fast configuration gives at the first item failure, or with CompletionReason::CustomCompletionFailed. When an item failed, durable::OperationErrorKind::ChildContext contains a child error whose kind is durable::ChildContextErrorKind::ChildFailed, and wire() holds the item's recorded failure. A custom failed outcome with no failed item has child kind BatchCompletionFailed and recorded type BatchCompletionError. Failure tolerance exceeded with no failed item has child kind Internal. .await_batch() returns these item and completion outcomes in the BatchResult instead. Both forms can return an OperationError for a task ownership violation. A panic in an item function propagates to the caller. See Error handling.

Task<IBatchResult<TResult>> MapAsync<TItem, TResult>(
    IReadOnlyList<TItem> items,
    Func<IDurableContext, TItem, int, IReadOnlyList<TItem>, CancellationToken, Task<TResult>> func,
    string? name = null,
    MapConfig<TItem>? config = null,
    CancellationToken cancellationToken = default);

Parameters:

items An IReadOnlyList<TItem> of items to process.
func A function called for each item. See Map Function.
name (optional) A name for the map operation. Omit it to infer one from the call site.
config (optional) A MapConfig<TItem> object.
cancellationToken (optional) A token linked with the SDK's workflow-shutdown signal, forwarded to func.

Returns: Task<IBatchResult<TResult>>. Use await to get the result.

Throws: Item exceptions are captured in the IBatchResult. Inspect Failed to detect failures, or call ThrowIfError() to re-throw the first failure. The map throws MapException only when the CompletionConfig criteria are violated.

Map Function¶

TypeScriptPythonJavaGoRustC#

type MapFunc<TInput, TOutput> = (
  context: DurableContext,
  item: TInput,
  index: number,
  array: TInput[],
) => Promise<TOutput>

Parameters:

context The child DurableContext for this item's execution.
item The current item being processed.
index The zero-based index of the item in the input array.
array The full input array.

Returns: Promise<TOutput>.

Callable[[DurableContext, T, int, Sequence[T]], R]

Parameters:

ctx The child DurableContext for this item's execution.
item The current item being processed.
index The zero-based index of the item in the input sequence.
items The full input sequence.

Returns: R.

@FunctionalInterface
interface MapFunction<I, O> {
    O apply(I item, int index, DurableContext context);
}

Parameters:

item The current item being processed.
index The zero-based index of the item in the input collection.
context The child DurableContext for this item's execution.

Returns: O.

The item function is the fn parameter of durable.Map, with the signature func(ctx durable.Context, item I, index int) (O, error).

Parameters:

ctx The child durable.Context for this item's execution.
item The current item.
index The zero-based index of the item.

Returns: (O, error).

The function takes no fourth argument for the full collection. The slice is in scope at the call site, so close over it when an item depends on the rest of the collection.

The item function is the f parameter of map, with these bounds:

F: Fn(DurableContext, I, usize) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<O, BoxError>> + Send + 'static,

Parameters:

DurableContext The child context for this item's execution.
I The current item, moved into the call.
usize The zero-based index of the item.

Returns: a future of Result<O, durable::BoxError>. The ? operator converts the error of a durable operation inside the function into durable::BoxError. With the default durable::JsonSerdes, O must implement serde::Serialize and serde::de::DeserializeOwned.

Pass a closure or an async fn with this signature. To read other items in the function, clone the collection into an Arc<[I]> before the call and capture a clone of it in the closure.

Func<IDurableContext, TItem, int, IReadOnlyList<TItem>, CancellationToken, Task<TResult>>

Parameters:

context The child IDurableContext for this item's execution.
item The current item being processed.
index The zero-based index of the item in the input list.
items The full input list.
cancellationToken A token linked with the SDK's workflow-shutdown signal. It is also tripped when a sibling item satisfies the CompletionConfig and the map short-circuits.

Returns: Task<TResult>.

MapConfig¶

TypeScriptPythonJavaGoRustC#

interface MapConfig<TItem, TResult> {
  maxConcurrency?: number;
  itemNamer?: (item: TItem, index: number) => string;
  completionConfig?: CompletionConfig;
  serdes?: Serdes<BatchResult<TResult>>;
  itemSerdes?: Serdes<TResult>;
  summaryGenerator?: (result: BatchResult<TResult>) => string;
  nesting?: NestingType;
}

Parameters:

maxConcurrency (optional) Maximum items running at once. Default: unlimited.
itemNamer (optional) A function that returns a custom name for each item, used in logs and tests.
completionConfig (optional) When to stop. Default: wait for all items.
serdes (optional) Custom Serdes for the BatchResult.
itemSerdes (optional) Custom Serdes for individual item results.
summaryGenerator (optional) A function invoked when the serialized BatchResult exceeds 256KB. See Checkpointing.
nesting (optional) NestingType.NESTED (default) or NestingType.FLAT. See Nesting.

@dataclass(frozen=True)
class MapConfig(Generic[T]):
    max_concurrency: int | None = None
    completion_config: CompletionConfig = CompletionConfig()
    serdes: SerDes | None = None
    item_serdes: SerDes | None = None
    summary_generator: SummaryGenerator | None = None
    nesting_type: NestingType = NestingType.NESTED
    item_namer: Callable[[T, int], str] | None = None

Parameters:

max_concurrency (optional) Maximum items running at once. Default: unlimited.
completion_config (optional) When to stop. Default: CompletionConfig() (lenient, all items run regardless of failures).
serdes (optional) Custom SerDes for the BatchResult.
item_serdes (optional) Custom SerDes for individual item results.
summary_generator (optional) A callable invoked when the serialized BatchResult exceeds 256KB. See Checkpointing.
nesting_type (optional) NestingType.NESTED (default) or NestingType.FLAT. See Nesting.
item_namer (optional) A deterministic callable that returns a custom name for each item from the item and its zero-based index.

MapConfig.builder()
    .maxConcurrency(Integer)       // optional
    .completionConfig(CompletionConfig)  // optional
    .serDes(SerDes)                // optional
    .nestingType(NestingType)      // optional
    .itemNamer(BiFunction<Object, Integer, String>)  // optional
    .itemNamer(Class<I>, BiFunction<? super I, Integer, String>)  // optional
    .build()

Parameters:

maxConcurrency (optional) Maximum items running at once. Default: unlimited.
completionConfig (optional) When to stop. Default: CompletionConfig.allCompleted().
serDes (optional) Custom SerDes for item results and the overall result.
nestingType (optional) NestingType.NESTED (default) or NestingType.FLAT. See Nesting.
itemNamer (optional) A function that returns a custom name for each item from the item and its zero-based index. Pass the item type as the first argument to receive the item strongly typed instead of as Object. Java does not support itemNamer with NestingType.FLAT.

Configure Map with BatchOption values passed as trailing arguments. The same options configure Parallel.

func WithMaxConcurrency(n int) BatchOption
func WithCompletion(c CompletionConfig) BatchOption
func WithItemNamer(namer func(index int) string) BatchOption
func WithNesting(m NestingMode) BatchOption
func WithBatchSummary[O any](fn func(result BatchResult[O]) string) BatchOption
func WithBatchSerdes(s Serdes) BatchOption
func WithBatchResultSerdes(s Serdes) BatchOption

Parameters:

WithMaxConcurrency Maximum items running at once. Zero or negative is invalid. Default: unlimited.
WithCompletion When to stop. Default: fail-fast. See CompletionConfig.
WithItemNamer Names each item from its zero-based index. The namer receives only the index. An index whose namer returns "" is named map-item-<index>. The namer must be deterministic.
WithNesting NestingNormal (default) or NestingFlat. See Nesting.
WithBatchSummary A summary for an oversized result. See Checkpointing.
WithBatchSerdes Serializer for each item result.
WithBatchResultSerdes Serializer for the whole BatchResult.

Configure map with methods on the MapBuilder it returns. Call them before you await the builder. The same methods, except item_namer, configure parallel.

impl<I, O, F, Fut, IS, RS> MapBuilder<I, O, F, Fut, IS, RS>
...
    pub fn name(mut self, name: impl Into<String>) -> Self
    ...
    pub fn max_concurrency(mut self, max: usize) -> Self
    ...
    pub fn completion(mut self, config: CompletionConfig) -> Self
    ...
    pub fn serdes<IS2>(self, serdes: IS2) -> MapBuilder<I, O, F, Fut, IS2, RS>
    where
        IS2: Serdes<O>,
    ...
    pub fn result_serdes<RS2>(self, serdes: RS2) -> MapBuilder<I, O, F, Fut, IS, RS2>
    where
        RS2: Serdes<String>,
    ...
    pub fn nesting(mut self, mode: NestingMode) -> Self
    ...
    pub fn item_namer(
        mut self,
        namer: impl Fn(&I, usize) -> String + Send + Sync + 'static,
    ) -> Self
    ...
    pub fn summary(
        mut self,
        summary: impl Fn(&BatchResult<O>) -> String + Send + Sync + 'static,
    ) -> Self
...

pub enum NestingMode {
    Normal,
    Flat,
}

NestingMode is non-exhaustive, so a match needs a _ arm.

Parameters:

name A name for the map operation. Default: unnamed.
max_concurrency Maximum items running at once. Default: unlimited. 0 fails the execution with an Invalid max_concurrency: 0 error, which the handler cannot catch.
completion When to stop. Default: fail-fast. See CompletionConfig.
serdes A custom durable::Serdes<O> for individual item results. Default: durable::JsonSerdes. See Serialization.
result_serdes A custom durable::Serdes<String> for the whole batch record, which it receives as JSON text. Its deserialize must return the exact text that serialize received. Default: durable::PassthroughSerdes, which stores the text unchanged.
nesting NestingMode::Normal (default) or NestingMode::Flat. See Nesting.
item_namer A function that returns a custom name for each item from a reference to the item and its zero-based index. See Naming map operations.
summary A function invoked when the serialized batch result exceeds 256KB. See Checkpointing.

CompletionConfig, NestingMode, and the result types are in the durable::builders::map_parallel module.

public sealed class MapConfig<TItem>
{
    public int? MaxConcurrency { get; set; }               // null = unlimited
    public CompletionConfig CompletionConfig { get; set; } // default AllSuccessful()
    public NestingType NestingType { get; set; }           // default Nested
    public Func<TItem, int, string>? ItemNamer { get; set; }
}

Parameters:

MaxConcurrency (optional) Maximum items running at once. null (default) is unlimited; must be at least 1 when set.
CompletionConfig (optional) When to stop. Default: CompletionConfig.AllSuccessful(). Any item failure completes the map with FailureToleranceExceeded. Set CompletionConfig.AllCompleted() to run every item regardless of failures.
NestingType (optional) NestingType.Nested (default) or NestingType.Flat. See Nesting.
ItemNamer (optional) A function that returns a custom name for each item, given the item and its zero-based index. Used in logs and traces. When null (default), items are named by index.

The BatchResult and per-item results are serialized with the ILambdaSerializer registered on ILambdaContext.Serializer; there is no per-item serializer slot. See Serialization.

CompletionConfig¶

See Completion strategies for how CompletionConfig affects execution and the completion status of the result.

TypeScriptPythonJavaGoRustC#

interface CompletionConfig {
  minSuccessful?: number;
  toleratedFailureCount?: number;
  toleratedFailurePercentage?: number;
}

@dataclass(frozen=True)
class CompletionConfig:
    min_successful: int | None = None
    tolerated_failure_count: int | None = None
    tolerated_failure_percentage: int | float | None = None
    should_complete: Callable[[CompletionStatus], CompletionDecision] | None = None

Use the threshold fields for count-based rules, or should_complete for a custom predicate (see Completion strategies). should_complete cannot be combined with the threshold fields.

CompletionConfig.allCompleted()
CompletionConfig.allSuccessful()
CompletionConfig.firstSuccessful()
CompletionConfig.minSuccessful(int count)
CompletionConfig.toleratedFailureCount(int count)
CompletionConfig.toleratedFailurePercentage(double percentage)
CompletionConfig.shouldComplete(
    Function<CompletionStatus, CompletionDecision> decision)

Set the fields of a CompletionConfig and pass it with WithCompletion. The threshold fields and ShouldComplete are mutually exclusive.

type CompletionConfig struct {
    MinSuccessful              int
    ToleratedFailureCount      *int
    ToleratedFailurePercentage *int
    ShouldComplete             func(BatchProgress) CompletionDecision
}

type BatchProgress struct {
    TotalCount     int
    CompletedCount int
    SuccessCount   int
    FailureCount   int
    Items          []BatchItemProgress
}

type BatchItemProgress struct {
    Index  int
    Name   string
    Status BatchItemStatus
}

func ContinueBatch() CompletionDecision
func CompleteBatch(outcome CompletionOutcome) CompletionDecision

type CompletionOutcome int

const (
    CompletionOutcomeSucceeded CompletionOutcome = 1
    CompletionOutcomeFailed    CompletionOutcome = 2
)

MinSuccessful Completes the batch once this many items succeed. Zero leaves it unset.
ToleratedFailureCount Fails the batch once more than this many items fail. Nil leaves it unset. aws.Int(0) fails on the first failure.
ToleratedFailurePercentage Fails the batch once the failure percentage, computed against the total item count, is strictly greater than this value. Nil leaves it unset. aws.Int(0) fails on the first failure.
ShouldComplete A custom predicate. It cannot be combined with the threshold fields.

Set the pointer fields with any *int, for example aws.Int from github.com/aws/aws-sdk-go-v2/aws. The default (no WithCompletion) is fail-fast. For all-completed, set ToleratedFailurePercentage: aws.Int(100), which is never exceeded. For first-successful, set MinSuccessful: 1.

Build a CompletionConfig with one of its constructors, or with CompletionConfig::builder() to set several thresholds. Pass it with .completion(...).

impl CompletionConfig {
    pub fn builder() -> CompletionConfigBuilder
    ...
    pub fn with_min_successful(min: usize) -> Self
    ...
    pub fn with_tolerated_failure_count(count: usize) -> Self
    ...
    pub fn with_tolerated_failure_percentage(
        pct: f64,
    ) -> Result<Self, CompletionConfigValidationError>
    ...
    pub fn with_completion_predicate<F>(predicate: F) -> Self
    where
        F: Fn(&BatchStats<'_>) -> CompletionDecision + Send + Sync + 'static,
    ...
}

impl CompletionConfigBuilder {
    ...
    pub fn min_successful(mut self, min: usize) -> Self
    ...
    pub fn tolerated_failure_count(mut self, count: usize) -> Self
    ...
    pub fn tolerated_failure_percentage(mut self, pct: f64) -> Self
    ...
    pub fn completion_predicate<F>(mut self, predicate: F) -> Self
    where
        F: Fn(&BatchStats<'_>) -> CompletionDecision + Send + Sync + 'static,
    ...
    pub fn build(self) -> Result<CompletionConfig, CompletionConfigValidationError>
    ...
}

pub enum CompletionDecision {
    Continue,
    Complete(CompletionOutcome),
}

impl CompletionDecision {
    pub fn complete(outcome: CompletionOutcome) -> Self
    ...
}

pub enum CompletionOutcome {
    Succeeded,
    Failed,
}

CompletionDecision and CompletionOutcome are non-exhaustive, so a match needs a _ arm.

min_successful Completes the batch once this many items succeed. On its own it tolerates every failure.
tolerated_failure_count Fails the batch once more than this many items fail. 0 fails on the first failure.
tolerated_failure_percentage Fails the batch once the failure percentage, computed against the total item count, is strictly greater than this value. The value must be finite and from 0.0 to 100.0. For any other value, with_tolerated_failure_percentage(...) and build() return a CompletionConfigValidationError, so an invalid value never reaches the map.
completion_predicate A custom predicate. build() returns an error when it is combined with a threshold. See Completion strategies.

When several thresholds are set, the first one reached ends the batch. The default (no .completion(...)) is fail-fast. For all-completed, use CompletionConfig::with_tolerated_failure_percentage(100.0)?, which is never exceeded. For first-successful, use CompletionConfig::with_min_successful(1).

Use the static factories or set the properties directly:

CompletionConfig.AllSuccessful()   // default for map: ToleratedFailureCount = 0
CompletionConfig.AllCompleted()    // every item runs regardless of failures
CompletionConfig.FirstSuccessful() // MinSuccessful = 1

new CompletionConfig { MinSuccessful = count }
new CompletionConfig { ToleratedFailureCount = count }
new CompletionConfig { ToleratedFailurePercentage = ratio } // ratio in [0.0, 1.0]

Result types¶

TypeScriptPythonJavaGoRustC#

Map returns the same BatchResult<TResult> type as parallel.

interface BatchResult<TResult> {
  all: BatchItem<TResult>[];
  status: BatchItemStatus.SUCCEEDED | BatchItemStatus.FAILED;
  completionReason: "ALL_COMPLETED" | "MIN_SUCCESSFUL_REACHED" | "FAILURE_TOLERANCE_EXCEEDED";
  hasFailure: boolean;
  successCount: number;
  failureCount: number;
  startedCount: number;
  totalCount: number;
  getResults(): TResult[];
  getErrors(): ChildContextError[];
  succeeded(): BatchItem<TResult>[];
  failed(): BatchItem<TResult>[];
  started(): BatchItem<TResult>[];
  throwIfError(): void;
}

all all BatchItem entries, one per item, in input order
getResults() results of succeeded items, preserving input order
getErrors() ChildContextError[] for failed items
succeeded() / failed() / started() BatchItem[] filtered by status
successCount / failureCount / startedCount / totalCount item counts
status SUCCEEDED if no failures, FAILED otherwise
completionReason why the operation completed. See Completion strategies.
hasFailure true if any item failed
throwIfError() throws the first item error, if any

interface BatchItem<TResult> {
  index: number;
  status: BatchItemStatus;
  result?: TResult;
  error?: ChildContextError;
}

enum BatchItemStatus {
  SUCCEEDED = "SUCCEEDED",
  FAILED    = "FAILED",
  STARTED   = "STARTED",
}

Map returns the same BatchResult[R] type as parallel.

@dataclass(frozen=True)
class BatchResult(Generic[R]):
    all: list[BatchItem[R]]
    completion_reason: CompletionReason

    def get_results(self) -> list[R]: ...
    def get_errors(self) -> list[ErrorObject]: ...
    def succeeded(self) -> list[BatchItem[R]]: ...
    def failed(self) -> list[BatchItem[R]]: ...
    def started(self) -> list[BatchItem[R]]: ...
    def throw_if_error(self) -> None: ...
    def to_dict(self) -> dict: ...

    @property
    def status(self) -> BatchItemStatus: ...
    @property
    def has_failure(self) -> bool: ...
    @property
    def success_count(self) -> int: ...
    @property
    def failure_count(self) -> int: ...
    @property
    def started_count(self) -> int: ...
    @property
    def total_count(self) -> int: ...

all all BatchItem entries, one per item, in input order
get_results() results of succeeded items, preserving input order
get_errors() list[ErrorObject] for failed items
succeeded() / failed() / started() BatchItem lists filtered by status
success_count / failure_count / started_count / total_count item counts
status BatchItemStatus.SUCCEEDED if no failures, FAILED otherwise
completion_reason why the operation completed. See Completion strategies.
has_failure True if any item failed
throw_if_error() raises the first failure: ChildContextError for a failed item, SerDesError if an item result failed to serialize, or BatchCompletionError if a custom predicate failed the batch
to_dict() serializes to a plain dict. Serializability depends on R.

Map returns MapResult<O>, which differs from ParallelResult. It holds per-item results with individual status, result, and error fields.

record MapResult<T>(
    List<MapResultItem<T>> items,
    ConcurrencyCompletionStatus completionReason
) {
    MapResultItem<T> getItem(int index)
    T getResult(int index)
    MapError getError(int index)
    boolean allSucceeded()
    int size()
    List<T> results()        // all results, nulls for failed/skipped items
    List<T> succeeded()      // results of succeeded items only
    List<MapError> failed()  // errors of failed items only
}

record MapResultItem<T>(Status status, T result, MapError error) {
    enum Status { SUCCEEDED, FAILED, SKIPPED }
}

record MapError(String errorType, String errorMessage, List<String> stackTrace) {}

enum ConcurrencyCompletionStatus {
    ALL_COMPLETED,
    MIN_SUCCESSFUL_REACHED,
    FAILURE_TOLERANCE_EXCEEDED,
    CUSTOM_COMPLETION_SUCCEEDED,
    CUSTOM_COMPLETION_FAILED
}

items ordered list of MapResultItem, one per input item
getItem(index) the MapResultItem at the given index
getResult(index) the result at the given index, or null if failed or skipped
getError(index) the MapError at the given index, or null if succeeded or skipped
allSucceeded() true if every item has status SUCCEEDED
size() total number of items
results() all results as a list, with null for failed or skipped items
succeeded() results of items with status SUCCEEDED
failed() MapError objects for items with status FAILED
completionReason why the operation completed. See Completion strategies.

Items that did not start before the operation reached its completion criteria have status SKIPPED (not STARTED as in TypeScript and Python).

Map and Parallel return the same BatchResult[O].

type BatchResult[O any] struct {
    Items  []BatchItem[O]
    Reason CompletionReason
}

func (r BatchResult[O]) Results() []O
func (r BatchResult[O]) Succeeded() []BatchItem[O]
func (r BatchResult[O]) Failed() []BatchItem[O]
func (r BatchResult[O]) Started() []BatchItem[O]
func (r BatchResult[O]) Errors() []error
func (r BatchResult[O]) HasFailure() bool
func (r BatchResult[O]) SuccessCount() int
func (r BatchResult[O]) FailureCount() int
func (r BatchResult[O]) StartedCount() int
func (r BatchResult[O]) TotalCount() int
func (r BatchResult[O]) Status() BatchItemStatus
func (r BatchResult[O]) Item(name string) *BatchItem[O]
func (r BatchResult[O]) Result(name string) (value O, ok bool)

type BatchItem[O any] struct {
    Index  int
    Name   string
    Status BatchItemStatus
    Result O
    Err    error
}

type BatchItemStatus int

const (
    BatchItemNotStarted BatchItemStatus = 0
    BatchItemSucceeded  BatchItemStatus = 1
    BatchItemFailed     BatchItemStatus = 2
    BatchItemStarted    BatchItemStatus = 4
)

type CompletionReason int

const (
    CompletionAllCompleted             CompletionReason = 1
    CompletionMinSuccessfulReached     CompletionReason = 2
    CompletionFailureToleranceExceeded CompletionReason = 3
    CompletionCustomSucceeded          CompletionReason = 4
    CompletionCustomFailed             CompletionReason = 5
)

Items per-item outcomes in input order. An item started and then abandoned on early completion is included with BatchItemStarted. Items that never started are omitted.
Results() successful results in input order.
Errors() errors of failed items, in input order. Each is rebuilt from its checkpoint record. In NestingNormal each is a *durable.ChildContextError named after the item. In NestingFlat each is the rebuilt error the item function returned. errors.As matches an SDK error inside, such as *durable.StepError, but not your own error types. In NestingNormal, match your own types on ChildContextError.ErrorType.
Succeeded() / Failed() / Started() items filtered by status.
SuccessCount() / FailureCount() / StartedCount() / TotalCount() item counts. TotalCount() excludes items that never started.
Status() BatchItemFailed if any item failed or the batch failed as a unit, else BatchItemSucceeded. A custom decision sets it directly. A failure within tolerance still gives BatchItemFailed with a nil error.
Item(name) / Result(name) look up one item by its recorded name.
Reason why the batch completed.

BatchItemStatus.String() and CompletionReason.String() return the wire forms, such as SUCCEEDED and ALL_COMPLETED.

Inspect the returned error and the result. Each BatchItem holds Result (set when Status is BatchItemSucceeded) and Err (set when Status is BatchItemFailed).

.await_batch() returns a BatchResult<O>, the same type that parallel returns.

pub struct BatchResult<O> {
    pub items: Vec<BatchItem<O>>,
    pub reason: CompletionReason,
}

impl<O> BatchResult<O> {
    ...
    pub fn results(&self) -> impl Iterator<Item = &O> + '_
    ...
    pub fn success_count(&self) -> usize
    ...
    pub fn failure_count(&self) -> usize
    ...
    pub fn has_failure(&self) -> bool
    ...
    pub fn status(&self) -> BatchStatus
    ...
    pub fn total_count(&self) -> usize
    ...
    pub fn errors(&self) -> impl Iterator<Item = BatchError<'_>> + '_
    ...
    pub fn succeeded(&self) -> impl Iterator<Item = &BatchItem<O>> + '_
    ...
    pub fn failed(&self) -> impl Iterator<Item = &BatchItem<O>> + '_
    ...
    pub fn started(&self) -> impl Iterator<Item = &BatchItem<O>> + '_
    ...
    pub fn started_count(&self) -> usize
    ...
    pub fn into_result(self) -> Result<Self, OperationError>
    ...
}
...

pub enum BatchItemStatus {
    Succeeded,
    Failed,
    Started,
}

pub struct BatchItem<O> {
    pub index: usize,
    pub name: String,
    pub status: BatchItemStatus,
    pub result: Option<O>,
    pub error_message: Option<String>,
    pub error_type: Option<String>,
    pub error_data: Option<String>,
    pub stack_trace: Vec<String>,
}
...

pub enum BatchStatus {
    Succeeded,
    Failed,
}
...

pub struct BatchError<'a> {
    pub index: usize,
    pub name: &'a str,
    pub message: &'a str,
    pub error_type: Option<&'a str>,
}

pub enum CompletionReason {
    AllCompleted,
    MinSuccessfulReached,
    FailureToleranceExceeded,
    CustomCompletionSucceeded,
    CustomCompletionFailed,
}

The structs and enums in this block are non-exhaustive. Enum matches need a _ arm, and struct patterns need ...

items one BatchItem per started item, in input order. Items that never started are omitted.
reason why the operation completed. See Completion strategies.
results() an iterator over references to the results of succeeded items, in input order
errors() an iterator with a BatchError for each failed item, with its index, name, message, and error type
succeeded() / failed() / started() iterators over the items with that status
success_count() / failure_count() / started_count() / total_count() item counts. total_count() excludes items that never started.
status() the outcome of a custom predicate when one completed the batch. Otherwise BatchStatus::Failed if any item failed, also within the failure tolerance, and BatchStatus::Succeeded if none failed.
has_failure() true if any item failed
into_result() returns the first failed item's ChildFailed error, if any. When no item failed and the status is Failed, it returns a BatchCompletionFailed error. Otherwise it returns the BatchResult. See Errors.

A BatchItem holds result when its status is BatchItemStatus::Succeeded, and the recorded failure (error_message, error_type, error_data, stack_trace) when it is BatchItemStatus::Failed. name is the item's recorded name. BatchItemStatus::Started marks an item that was running when the batch completed. CompletionReason::as_str() and BatchStatus::as_str() return the wire forms, such as ALL_COMPLETED and SUCCEEDED.

Map returns the same IBatchResult<TResult> type as parallel. It holds per-item results with individual status, result, and error.

public interface IBatchResult<T> : IBatchResult
{
    IReadOnlyList<IBatchItem<T>> All { get; }        // one per item, index order
    IReadOnlyList<IBatchItem<T>> Succeeded { get; }
    IReadOnlyList<IBatchItem<T>> Failed { get; }
    IReadOnlyList<IBatchItem<T>> Started { get; }
    IReadOnlyList<T> GetResults();                   // succeeded results, index order
    IReadOnlyList<DurableExecutionException> GetErrors();
    void ThrowIfError();                             // throws first item error, if any
}

public interface IBatchResult
{
    CompletionReason CompletionReason { get; }
    bool HasFailure { get; }
    int SuccessCount { get; }
    int FailureCount { get; }
    int StartedCount { get; }
    int TotalCount { get; }
}

All all IBatchItem entries, one per item, in original index order
GetResults() results of succeeded items, preserving index order
GetErrors() DurableExecutionException for failed items, in index order
Succeeded / Failed / Started IBatchItem lists filtered by status
SuccessCount / FailureCount / StartedCount / TotalCount item counts
CompletionReason why the operation completed. See Completion strategies.
HasFailure true if any item failed
ThrowIfError() throws the first item error, if any

public interface IBatchItem<T>
{
    int Index { get; }
    string? Name { get; }
    BatchItemStatus Status { get; }
    T? Result { get; }                       // set when Status == Succeeded
    DurableExecutionException? Error { get; } // set when Status == Failed
}

public enum BatchItemStatus
{
    Succeeded,
    Failed,
    Started
}

Items that did not start before the operation reached its completion criteria have status Started.

The map function¶

The map function can use any durable operation such as steps, waits, or nested map and parallel operations. Each item runs in its own child context, so items do not share state with each other or with the parent context.

TypeScriptPythonJavaGoRustC#

import { DurableContext } from "@aws/durable-execution-sdk-js";

type Order = { id: string; amount: number };
type Receipt = { orderId: string; charged: number };

async function processOrder(
  ctx: DurableContext,
  order: Order,
  index: number,
  orders: Order[],
): Promise<Receipt> {
  const validated = await ctx.step("validate", async () => {
    if (order.amount <= 0) throw new Error("Invalid amount");
    return order;
  });
  const charged = await ctx.step("charge", async () => validated.amount);
  return { orderId: validated.id, charged };
}

from aws_durable_execution_sdk_python import DurableContext


def process_order(
    ctx: DurableContext,
    order: dict,
    index: int,
    orders: list[dict],
) -> dict:
    def validate(_):
        if order["amount"] <= 0:
            raise ValueError("Invalid amount")
        return order

    validated = ctx.step(validate, name="validate")
    charged = ctx.step(lambda _: validated["amount"], name="charge")
    return {"orderId": validated["id"], "charged": charged}

import software.amazon.lambda.durable.DurableContext;

record Order(String id, double amount) {}
record Receipt(String orderId, double charged) {}

// MapFunction<Order, Receipt> implementation
Receipt processOrder(Order order, int index, DurableContext ctx) {
    var validated = ctx.step("validate", Order.class, s -> {
        if (order.amount() <= 0) throw new IllegalArgumentException("Invalid amount");
        return order;
    });
    var charged = ctx.step("charge", Double.class, s -> validated.amount());
    return new Receipt(validated.id(), charged);
}

// Package main shows the item function a Map applies to each item. The
// function runs steps in the item's own child context and returns a value.
package main

import (
    "errors"

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

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

type Receipt struct {
    OrderID string `json:"orderId"`
    Charged int    `json:"charged"`
}

type Input struct {
    Orders []Order `json:"orders"`
}

func processOrder(ctx durable.Context, order Order, index int) (Receipt, error) {
    validated, err := durable.Step(ctx, "validate", func(durable.StepContext) (Order, error) {
        if order.Amount <= 0 {
            return Order{}, errors.New("invalid amount")
        }
        return order, nil
    })
    if err != nil {
        return Receipt{}, err
    }
    charged, err := durable.Step(ctx, "charge", func(durable.StepContext) (int, error) {
        return validated.Amount, nil
    })
    if err != nil {
        return Receipt{}, err
    }
    return Receipt{OrderID: validated.ID, Charged: charged}, nil
}

func handler(ctx durable.Context, event Input) ([]Receipt, error) {
    result, err := durable.Map(ctx, "process-orders", event.Orders, processOrder)
    if err != nil {
        return nil, err
    }
    return result.Results(), nil
}

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

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

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

#[derive(Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct Receipt {
    order_id: String,
    charged: f64,
}

async fn process_order(
    ctx: durable::DurableContext,
    order: Order,
    _index: usize,
) -> Result<Receipt, durable::BoxError> {
    let validated = ctx
        .step(move |_| async move {
            if order.amount <= 0.0 {
                return Err("Invalid amount".into());
            }
            Ok(order)
        })
        .name("validate")
        .await?;
    let amount = validated.amount;
    let charged = ctx
        .step(move |_| async move { Ok(amount) })
        .name("charge")
        .await?;
    Ok(Receipt {
        order_id: validated.id,
        charged,
    })
}

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

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

    private async Task<IReadOnlyList<Receipt>> Workflow(
        IReadOnlyList<Order> orders, IDurableContext ctx)
    {
        IBatchResult<Receipt> result = await ctx.MapAsync(
            orders, ProcessOrder, name: "process-orders");
        return result.GetResults();
    }

    // The map function: receives (ctx, item, index, allItems, cancellationToken)
    private static async Task<Receipt> ProcessOrder(
        IDurableContext ctx, Order order, int index,
        IReadOnlyList<Order> orders, CancellationToken ct)
    {
        Order validated = await ctx.StepAsync(async (_, _) =>
        {
            if (order.Amount <= 0) throw new ArgumentException("Invalid amount");
            return order;
        }, name: "validate");

        decimal charged = await ctx.StepAsync(
            async (_, _) => validated.Amount, name: "charge");

        return new Receipt(validated.Id, charged);
    }
}

public record Order(string Id, decimal Amount);
public record Receipt(string OrderId, decimal Charged);

Naming map operations¶

Name your map operations to make them easier to identify in logs and tests.

TypeScriptPythonJavaGoRustC#

import {
  BatchResult,
  DurableContext,
  withDurableExecution,
} from "@aws/durable-execution-sdk-js";

export const handler = withDurableExecution(
  async (event: { userIds: string[] }, context: DurableContext): Promise<string[]> => {
    // Named: pass name as first argument, undefined to omit
    const result: BatchResult<string> = await context.map(
      "process-users",
      event.userIds,
      async (ctx, userId, index) =>
        ctx.step(`process-${index}`, async () => `processed-${userId}`),
    );

    return result.getResults();
  },
);

The name is the first argument. Pass undefined to omit it.

Use itemNamer in MapConfig to give each item a custom name:

context.map("process-orders", orders, processOrder, {
  itemNamer: (order, index) => `order-${order.id}`,
});

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)


def process_user(
    ctx: DurableContext, user_id: str, index: int, user_ids: list[str]
) -> str:
    return ctx.step(lambda _: f"processed-{user_id}", name=f"process-{index}")


@durable_execution
def handler(event: dict, context: DurableContext) -> list[str]:
    # Pass name as keyword argument; omit or pass None to leave unnamed
    result: BatchResult[str] = context.map(
        event["userIds"],
        process_user,
        name="process-users",
    )
    return result.get_results()

Pass name as a keyword argument. Omit it or pass None to leave it unnamed.

Use item_namer in MapConfig to give each item a custom name:

config = MapConfig(item_namer=lambda order, index: f"order-{order['id']}")

context.map(orders, process_order, name="process-orders", config=config)

import java.util.List;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.model.MapResult;

public class NamedMap extends DurableHandler<List<String>, List<String>> {
    @Override
    public List<String> handleRequest(List<String> userIds, DurableContext context) {
        // The name is always required in Java
        MapResult<String> result = context.map(
                "process-users",
                userIds,
                String.class,
                (userId, index, ctx) -> ctx.step(
                        "process-" + index, String.class, s -> "processed-" + userId));

        return result.succeeded();
    }
}

The name is always required in Java. The SDK derives each item's name from the operation name: {name}-iteration-{index}.

Use itemNamer in MapConfig to give each item a custom name:

var config = MapConfig.builder()
        .itemNamer(Order.class, (order, index) -> "order-" + order.id())
        .build();

context.map("process-orders", orders, ProcessedOrder.class, this::processOrder, config);

// Package main names a Map operation.
package main

import (
    "fmt"

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

type Input struct {
    UserIDs []string `json:"userIds"`
}

func handler(ctx durable.Context, event Input) ([]string, error) {
    result, err := durable.Map(ctx, "process-users", event.UserIDs,
        func(ctx durable.Context, userID string, index int) (string, error) {
            return durable.Step(ctx, fmt.Sprintf("process-%d", index),
                func(durable.StepContext) (string, error) {
                    return "processed-" + userID, nil
                })
        })
    if err != nil {
        return nil, err
    }
    return result.Results(), nil
}

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

The name is the required second argument. Pass "" to leave it unnamed.

Use WithItemNamer to give each item a custom name. The namer receives only the item's zero-based index, so close over the input slice to name from the item value:

durable.Map(ctx, "process-orders", orders, processOrder,
    durable.WithItemNamer(func(i int) string { return "order-" + orders[i].ID }))

The namer must be deterministic. An index whose namer returns "" is named map-item-<index>.

use aws_durable_execution_sdk as durable;
use serde::Deserialize;

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct UsersEvent {
    user_ids: Vec<String>,
}

async fn handler(
    event: UsersEvent,
    ctx: durable::DurableContext,
) -> Result<Vec<String>, durable::BoxError> {
    let results = ctx
        .map(event.user_ids, |ctx, user_id, index| async move {
            Ok(ctx
                .step(move |_| async move { Ok(format!("processed-{user_id}")) })
                .name(format!("process-{index}"))
                .await?)
        })
        .name("process-users")
        .await?;

    Ok(results)
}

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

Set the name with .name(...). Omit the call to leave the map unnamed.

Use .item_namer(...) to give each item a custom name:

let receipts = ctx
    .map(orders, process_order)
    .name("process-orders")
    .item_namer(|order, _index| format!("order-{}", order.id))
    .await?;

The namer receives a reference to the item and its zero-based index. It runs for every item when the map starts, and again on every replay, so it must be deterministic. Without a namer, or when the namer returns "", the item is named map-item-<index>.

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

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

    private async Task<IReadOnlyList<string>> Workflow(
        UserEvent input, IDurableContext ctx)
    {
        // The name is the optional trailing argument; omit it to leave the map unnamed
        IBatchResult<string> result = await ctx.MapAsync(
            input.UserIds,
            async (itemCtx, userId, index, userIds, ct) =>
                await itemCtx.StepAsync(
                    async (_, _) => $"processed-{userId}",
                    name: $"process-{index}"),
            name: "process-users");

        return result.GetResults();
    }
}

public record UserEvent(IReadOnlyList<string> UserIds);

The name is the optional trailing argument. Omit it to infer one from the call site.

Use ItemNamer in MapConfig to give each item a custom name:

var config = new MapConfig<Order>
{
    ItemNamer = (order, index) => $"order-{order.Id}",
};

Configuration¶

Configure map behavior using MapConfig:

TypeScriptPythonJavaGoRustC#

import {
  BatchResult,
  DurableContext,
  NestingType,
  withDurableExecution,
} from "@aws/durable-execution-sdk-js";

export const handler = withDurableExecution(
  async (event: { urls: string[] }, context: DurableContext): Promise<string[]> => {
    const result: BatchResult<string> = await context.map(
      "fetch-urls",
      event.urls,
      async (ctx, url, index) =>
        ctx.step(`fetch-${index}`, async () => {
          const response = await fetch(url);
          return response.text();
        }),
      {
        maxConcurrency: 5,
        completionConfig: { toleratedFailureCount: 2 },
        nesting: NestingType.FLAT,
      },
    );

    return result.getResults();
  },
);

import urllib.request

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)
from aws_durable_execution_sdk_python.config import (
    CompletionConfig,
    MapConfig,
    NestingType,
)


def fetch_url(
    ctx: DurableContext, url: str, index: int, urls: list[str]
) -> str:
    def do_fetch(_):
        with urllib.request.urlopen(url) as response:
            return response.read().decode()

    return ctx.step(do_fetch, name=f"fetch-{index}")


@durable_execution
def handler(event: dict, context: DurableContext) -> list[str]:
    config = MapConfig(
        max_concurrency=5,
        completion_config=CompletionConfig(tolerated_failure_count=2),
        nesting_type=NestingType.FLAT,
    )
    result: BatchResult[str] = context.map(
        event["urls"],
        fetch_url,
        name="fetch-urls",
        config=config,
    )
    return result.get_results()

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.List;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.config.CompletionConfig;
import software.amazon.lambda.durable.config.MapConfig;
import software.amazon.lambda.durable.config.NestingType;
import software.amazon.lambda.durable.model.MapResult;

public class MapConfigExample extends DurableHandler<List<String>, List<String>> {
    private static final HttpClient HTTP = HttpClient.newHttpClient();

    @Override
    public List<String> handleRequest(List<String> urls, DurableContext context) {
        var config = MapConfig.builder()
                .maxConcurrency(5)
                .completionConfig(CompletionConfig.toleratedFailureCount(2))
                .nestingType(NestingType.FLAT)
                .build();

        MapResult<String> result = context.map(
                "fetch-urls",
                urls,
                String.class,
                (url, index, ctx) -> ctx.step("fetch-" + index, String.class, s -> {
                    var request = HttpRequest.newBuilder(URI.create(url)).build();
                    return HTTP.sendAsync(request, HttpResponse.BodyHandlers.ofString())
                            .join()
                            .body();
                }),
                config);

        return result.succeeded();
    }
}

// Package main configures a Map with bounded concurrency, a failure
// tolerance, and flat nesting.
package main

import (
    "fmt"

    "github.com/aws/aws-durable-execution-sdk-go/durable"
    "github.com/aws/aws-sdk-go-v2/aws"
)

type Input struct {
    URLs []string `json:"urls"`
}

func handler(ctx durable.Context, event Input) ([]string, error) {
    result, err := durable.Map(ctx, "fetch-urls", event.URLs,
        func(ctx durable.Context, url string, index int) (string, error) {
            return durable.Step(ctx, fmt.Sprintf("fetch-%d", index),
                func(durable.StepContext) (string, error) {
                    return fetch(url), nil
                })
        },
        durable.WithMaxConcurrency(5),
        durable.WithCompletion(durable.CompletionConfig{ToleratedFailureCount: aws.Int(2)}),
        durable.WithNesting(durable.NestingFlat),
    )
    if err != nil {
        return nil, err
    }
    return result.Results(), nil
}

// fetch is a deterministic stand-in for an HTTP GET.
func fetch(url string) string {
    return "body of " + url
}

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

Set the options with methods on the builder that ctx.map returns. See MapConfig.

use aws_durable_execution_sdk as durable;
use durable::builders::map_parallel::{CompletionConfig, NestingMode};
use serde::Deserialize;

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

async fn handler(
    event: UrlsEvent,
    ctx: durable::DurableContext,
) -> Result<Vec<String>, durable::BoxError> {
    let results = ctx
        .map(event.urls, |ctx, url, index| async move {
            Ok(ctx
                .step(move |_| async move { Ok(fetch(&url)) })
                .name(format!("fetch-{index}"))
                .await?)
        })
        .name("fetch-urls")
        .max_concurrency(5)
        .completion(CompletionConfig::with_tolerated_failure_count(2))
        .nesting(NestingMode::Flat)
        .await?;

    Ok(results)
}

// A stand-in for an HTTP GET that returns the response body.
fn fetch(url: &str) -> String {
    format!("body of {url}")
}

#[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 MapConfigExample
{
    private static readonly HttpClient Http = new();

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

    private async Task<IReadOnlyList<string>> Workflow(
        UrlEvent input, IDurableContext ctx)
    {
        var config = new MapConfig<string>
        {
            MaxConcurrency = 5,
            CompletionConfig = new CompletionConfig { ToleratedFailureCount = 2 },
            NestingType = NestingType.Flat,
        };

        IBatchResult<string> result = await ctx.MapAsync(
            input.Urls,
            async (itemCtx, url, index, urls, ct) =>
                await itemCtx.StepAsync(
                    async (_, stepCt) => await Http.GetStringAsync(url, stepCt),
                    name: $"fetch-{index}"),
            name: "fetch-urls",
            config: config);

        return result.GetResults();
    }
}

public record UrlEvent(IReadOnlyList<string> Urls);

Nesting¶

Nested mode is the default. The SDK records each item context as a separate CONTEXT operation and checkpoints the item result there. Each item appears separately in the execution history.

In flat mode, the SDK uses a virtual context for each item and omits the per-item CONTEXT operation. Durable operations inside the map function still checkpoint and appear as children of the map operation. The SDK records the item outcome with the parent map operation.

Use flat mode for maps with many items when each item performs few durable operations and you do not need each item represented separately in the execution history. Flat mode removes one checkpointed operation per item while preserving checkpoints for durable operations inside each item.

Completion strategies¶

CompletionConfig controls when the map operation completes. When the operation reaches the completion criteria, it abandons items that have not completed yet. The abandoned items will keep running in the background but cannot checkpoint their results after the parent completes. The SDK makes a best-effort attempt to cancel ongoing work in abandoned items, but cancellation is not guaranteed.

TypeScriptPythonJavaGoRustC#

The BatchResult's completionReason indicates the stop condition. Items that had not started yet do not appear in result.all. Items that had started but not completed appear with status STARTED.

`completionConfig`	Early exit `completionReason`	Full completion `completionReason`
`{}` or omitted	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`
`toleratedFailureCount=N`	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`
`toleratedFailurePercentage=N`	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`
`minSuccessful=N`	`MIN_SUCCESSFUL_REACHED`	`ALL_COMPLETED`

The BatchResult's completion_reason indicates the stop condition. Items that started but did not complete appear in result.all with status STARTED. Items that never started are omitted from result.all, so total_count counts only the items that appear.

`completion_config`	Early exit `completion_reason`	Full completion `completion_reason`
`CompletionConfig()` (default)	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`
`first_successful()`	`MIN_SUCCESSFUL_REACHED`	`ALL_COMPLETED`
`all_completed()`	n/a	`ALL_COMPLETED`
`tolerated_failure_count=N`	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`
`tolerated_failure_percentage=N`	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`
`min_successful=N`	`MIN_SUCCESSFUL_REACHED`	`ALL_COMPLETED`

The default CompletionConfig() is fail-fast: any item failure exceeds the tolerance and completes the batch early. Use CompletionConfig.all_completed() to run every item regardless of failures.

Set should_complete when the threshold fields cannot express the rule. The predicate receives a CompletionStatus (success_count, failure_count, completed_count, total_count, and items, a per-item tuple of CompletionItemStatus) and returns a CompletionDecision: continue_batch() to keep going, or complete_batch(outcome) to stop, where outcome defaults to CompletionOutcome.SUCCEEDED. A CompletionOutcome.FAILED outcome marks the whole batch failed, and throw_if_error() then raises BatchCompletionError even when no individual item failed.

from aws_durable_execution_sdk_python import (
    BatchResult,
    CompletionStatus,
    DurableContext,
    complete_batch,
    continue_batch,
    durable_execution,
)
from aws_durable_execution_sdk_python.config import CompletionConfig, MapConfig


def process_item(ctx: DurableContext, item: str, index: int, items: list[str]) -> str:
    return ctx.step(lambda _: item.upper(), name=f"process-{index}")


def stop_after_two_successes(status: CompletionStatus):
    if status.success_count >= 2:
        return complete_batch()
    return continue_batch()


@durable_execution
def handler(event: dict, context: DurableContext) -> list[str]:
    config = MapConfig(
        completion_config=CompletionConfig(should_complete=stop_after_two_successes),
    )
    result: BatchResult[str] = context.map(
        event["items"],
        process_item,
        name="process-items",
        config=config,
    )
    return result.get_results()

The predicate runs before any item is scheduled (completed_count == 0) and again on each terminal or suspension event, so it must handle the initial zero-progress snapshot. It cannot be combined with min_successful, tolerated_failure_count, or tolerated_failure_percentage, and must be deterministic, side-effect-free, and monotonic. Unscheduled items report status=None in items.

The MapResult's completionReason indicates the stop condition. Items that did not start before the operation completed have status SKIPPED.

`completionConfig`	Early exit `completionReason`	Full completion `completionReason`
`allCompleted()` (default)	n/a	`ALL_COMPLETED`
`allSuccessful()`	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`
`firstSuccessful()`	`MIN_SUCCESSFUL_REACHED`	`ALL_COMPLETED`
`minSuccessful(N)`	`MIN_SUCCESSFUL_REACHED`	`ALL_COMPLETED`
`toleratedFailureCount(N)`	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`
`toleratedFailurePercentage(p)`	`FAILURE_TOLERANCE_EXCEEDED`	`ALL_COMPLETED`

Use CompletionConfig.shouldComplete(...) when the predefined thresholds cannot express the completion rule. The SDK evaluates the function as completion state changes. It receives a CompletionStatus with successCount, failureCount, completedCount, totalCount, and allItemsRegistered. Map registers all items before processing begins.

Return CompletionDecision.continueExecution() to keep processing. Return CompletionDecision.complete(...) with CUSTOM_COMPLETION_SUCCEEDED or CUSTOM_COMPLETION_FAILED to stop and classify the result.

var completion = CompletionConfig.shouldComplete(status -> {
    if (status.successCount() >= requiredSuccesses) {
        return CompletionConfig.CompletionDecision.complete(
                ConcurrencyCompletionStatus.CUSTOM_COMPLETION_SUCCEEDED);
    }
    if (status.failureCount() >= failureLimit) {
        return CompletionConfig.CompletionDecision.complete(
                ConcurrencyCompletionStatus.CUSTOM_COMPLETION_FAILED);
    }
    return CompletionConfig.CompletionDecision.continueExecution();
});

A custom completion function is mutually exclusive with minSuccessful, toleratedFailureCount, and toleratedFailurePercentage. It must return a non-null decision. Keep it deterministic and free of side effects. Items that have not started when it completes have status SKIPPED. CUSTOM_COMPLETION_FAILED does not throw automatically. Inspect result.completionReason().isSucceeded() to distinguish the custom outcomes.

BatchResult.Reason records the stop condition. Items that never started are omitted from Items. An item that started but did not complete appears with BatchItemStarted. Map does not leave abandoned items running. An abandoned item stops at its next durable operation. Map returns only after every started item has stopped.

`CompletionConfig`	Early exit `Reason`	Full completion `Reason`
zero value (default, fail-fast)	`CompletionFailureToleranceExceeded`	`CompletionAllCompleted`
`ToleratedFailureCount: aws.Int(N)`	`CompletionFailureToleranceExceeded`	`CompletionAllCompleted`
`ToleratedFailurePercentage: aws.Int(N)`	`CompletionFailureToleranceExceeded`	`CompletionAllCompleted`
`MinSuccessful: N`	`CompletionMinSuccessfulReached`	`CompletionAllCompleted`
`ShouldComplete: ...`	`CompletionCustomSucceeded` or `CompletionCustomFailed`	`CompletionAllCompleted`

The default is fail-fast. Set ToleratedFailurePercentage: aws.Int(100) to run every item regardless of failures.

Set ShouldComplete when the threshold fields cannot express the rule. The predicate receives a BatchProgress snapshot before the first item and again after each item reaches a terminal state, so it must handle the initial zero-progress snapshot. It must be deterministic. Return ContinueBatch() to keep going, or CompleteBatch(durable.CompletionOutcomeSucceeded) or CompleteBatch(durable.CompletionOutcomeFailed) to stop and classify the result. With ShouldComplete set, an item failure does not stop the batch by itself. A failed custom outcome returns a *BatchError, or a *BatchCompletionError when no item failed. A predicate that panics fails the operation with an error that is not a *BatchError.

// Package main stops a Map early with a custom completion decision once two
// items succeed.
package main

import (
    "fmt"
    "strings"

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

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

func handler(ctx durable.Context, event Input) ([]string, error) {
    result, err := durable.Map(ctx, "process-items", event.Items,
        func(ctx durable.Context, item string, index int) (string, error) {
            return durable.Step(ctx, fmt.Sprintf("process-%d", index),
                func(durable.StepContext) (string, error) {
                    return strings.ToUpper(item), nil
                })
        },
        durable.WithCompletion(durable.CompletionConfig{
            ShouldComplete: func(p durable.BatchProgress) durable.CompletionDecision {
                if p.SuccessCount >= 2 {
                    return durable.CompleteBatch(durable.CompletionOutcomeSucceeded)
                }
                return durable.ContinueBatch()
            },
        }),
    )
    if err != nil {
        return nil, err
    }
    return result.Results(), nil
}

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

The BatchResult's reason field records the stop condition. Items that never started are omitted from items. An item that started but did not complete appears with BatchItemStatus::Started. The map does not leave abandoned items running. The SDK cancels each one at its next .await, and none of its later checkpoints is written.

`CompletionConfig`	Early exit `reason`	Full completion `reason`
none (default, fail-fast)	`FailureToleranceExceeded`	`AllCompleted`
`with_tolerated_failure_count(n)`	`FailureToleranceExceeded`	`AllCompleted`
`with_tolerated_failure_percentage(p)`	`FailureToleranceExceeded`	`AllCompleted`
`with_min_successful(n)`	`MinSuccessfulReached`	`AllCompleted`
`with_completion_predicate(...)`	`CustomCompletionSucceeded` or `CustomCompletionFailed`	custom reason if it returns `Complete`; otherwise `AllCompleted`

The default is fail-fast. Use CompletionConfig::with_tolerated_failure_percentage(100.0)? to run every item regardless of failures.

Use with_completion_predicate when the thresholds cannot express the rule. The predicate receives a BatchStats with succeeded(), failed(), settled(), total_items(), and items(), one CompletionItem per input item. Each item carries its zero-based index, caller-supplied name, and current status. The predicate returns a CompletionDecision. Return CompletionDecision::Continue to keep going, or CompletionDecision::complete(CompletionOutcome::Succeeded) or CompletionDecision::complete(CompletionOutcome::Failed) to stop and classify the result. With a predicate set, an item failure does not stop the batch by itself. A Failed outcome makes .await return a ChildFailed error for the first failed item, or a BatchCompletionFailed error when no item failed.

use aws_durable_execution_sdk as durable;
use durable::builders::map_parallel::{
    BatchStats, CompletionConfig, CompletionDecision, CompletionOutcome,
};
use serde::Deserialize;

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

async fn process_item(
    ctx: durable::DurableContext,
    item: String,
    index: usize,
) -> Result<String, durable::BoxError> {
    Ok(ctx
        .step(move |_| async move { Ok(item.to_uppercase()) })
        .name(format!("process-{index}"))
        .await?)
}

fn stop_after_two_successes(stats: &BatchStats<'_>) -> CompletionDecision {
    if stats.succeeded() >= 2 {
        return CompletionDecision::complete(CompletionOutcome::Succeeded);
    }
    CompletionDecision::Continue
}

async fn handler(
    event: ItemsEvent,
    ctx: durable::DurableContext,
) -> Result<Vec<String>, durable::BoxError> {
    let results = ctx
        .map(event.items, process_item)
        .name("process-items")
        .completion(CompletionConfig::with_completion_predicate(
            stop_after_two_successes,
        ))
        .await?;

    Ok(results)
}

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

The SDK calls the predicate before each item starts and after each item settles, so it must handle the initial zero-progress snapshot. It must be a pure function of the BatchStats it receives, and it must be monotonic. Once it returns Complete with an outcome, it must return the same decision for every later snapshot. It cannot be combined with the thresholds. An item that is waiting to start has None as its status in items().

The IBatchResult's CompletionReason indicates the stop condition. Items that were not dispatched before the operation completed have status Started.

`CompletionConfig`	Early exit `CompletionReason`	Full completion `CompletionReason`
`AllSuccessful()` (default)	`FailureToleranceExceeded`	`AllCompleted`
`AllCompleted()`	n/a	`AllCompleted`
`FirstSuccessful()`	`MinSuccessfulReached`	`AllCompleted`
`MinSuccessful = N`	`MinSuccessfulReached`	`AllCompleted`
`ToleratedFailureCount = N`	`FailureToleranceExceeded`	`AllCompleted`
`ToleratedFailurePercentage = ratio`	`FailureToleranceExceeded`	`AllCompleted`

Note

When using a minSuccessful strategy, failures do not trigger early exit. If all items fail before the success threshold is reached, the operation completes with ALL_COMPLETED.

TypeScriptPythonJavaGoRustC#

import {
  BatchResult,
  DurableContext,
  withDurableExecution,
} from "@aws/durable-execution-sdk-js";

export const handler = withDurableExecution(
  async (event: { items: string[] }, context: DurableContext): Promise<string[]> => {
    const result: BatchResult<string> = await context.map(
      "process-items",
      event.items,
      async (ctx, item, index) =>
        ctx.step(`process-${index}`, async () => item.toUpperCase()),
      {
        completionConfig: { minSuccessful: 3 },
      },
    );

    return result.getResults();
  },
);

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)
from aws_durable_execution_sdk_python.config import CompletionConfig, MapConfig


def process_item(
    ctx: DurableContext, item: str, index: int, items: list[str]
) -> str:
    return ctx.step(lambda _: item.upper(), name=f"process-{index}")


@durable_execution
def handler(event: dict, context: DurableContext) -> list[str]:
    config = MapConfig(
        completion_config=CompletionConfig(min_successful=3),
    )
    result: BatchResult[str] = context.map(
        event["items"],
        process_item,
        name="process-items",
        config=config,
    )
    return result.get_results()

import java.util.List;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.config.CompletionConfig;
import software.amazon.lambda.durable.config.MapConfig;
import software.amazon.lambda.durable.model.MapResult;

public class MapCompletionConfig extends DurableHandler<List<String>, List<String>> {
    @Override
    public List<String> handleRequest(List<String> items, DurableContext context) {
        var config = MapConfig.builder()
                .completionConfig(CompletionConfig.minSuccessful(3))
                .build();

        MapResult<String> result = context.map(
                "process-items",
                items,
                String.class,
                (item, index, ctx) -> ctx.step(
                        "process-" + index, String.class, s -> item.toUpperCase()),
                config);

        return result.succeeded();
    }
}

// Package main completes a Map once three items succeed.
package main

import (
    "fmt"
    "strings"

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

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

func handler(ctx durable.Context, event Input) ([]string, error) {
    result, err := durable.Map(ctx, "process-items", event.Items,
        func(ctx durable.Context, item string, index int) (string, error) {
            return durable.Step(ctx, fmt.Sprintf("process-%d", index),
                func(durable.StepContext) (string, error) {
                    return strings.ToUpper(item), nil
                })
        },
        durable.WithCompletion(durable.CompletionConfig{MinSuccessful: 3}),
    )
    if err != nil {
        return nil, err
    }
    return result.Results(), nil
}

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

use aws_durable_execution_sdk as durable;
use durable::builders::map_parallel::CompletionConfig;
use serde::Deserialize;

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

async fn handler(
    event: ItemsEvent,
    ctx: durable::DurableContext,
) -> Result<Vec<String>, durable::BoxError> {
    let results = ctx
        .map(event.items, |ctx, item, index| async move {
            Ok(ctx
                .step(move |_| async move { Ok(item.to_uppercase()) })
                .name(format!("process-{index}"))
                .await?)
        })
        .name("process-items")
        .completion(CompletionConfig::with_min_successful(3))
        .await?;

    Ok(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 CompletionConfigExample
{
    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<ItemEvent, IReadOnlyList<string>>(
            Workflow, input, context);

    private async Task<IReadOnlyList<string>> Workflow(
        ItemEvent input, IDurableContext ctx)
    {
        var config = new MapConfig<string>
        {
            CompletionConfig = new CompletionConfig { MinSuccessful = 3 },
        };

        IBatchResult<string> result = await ctx.MapAsync(
            input.Items,
            async (itemCtx, item, index, items, ct) =>
                await itemCtx.StepAsync(
                    async (_, _) => item.ToUpperInvariant(),
                    name: $"process-{index}"),
            name: "process-items",
            config: config);

        return result.GetResults();
    }
}

public record ItemEvent(IReadOnlyList<string> Items);

Error handling¶

When an item throws an error, map captures the error in the result rather than propagating it immediately. Other items continue running.

TypeScriptPythonJavaGoRustC#

BatchResult.status is FAILED if any item failed. Call throwIfError() to propagate the first item error as an exception, or inspect getErrors() to handle errors individually.

import {
  BatchResult,
  DurableContext,
  withDurableExecution,
} from "@aws/durable-execution-sdk-js";

export const handler = withDurableExecution(
  async (event: { items: string[] }, context: DurableContext): Promise<void> => {
    const result: BatchResult<string> = await context.map(
      "process-items",
      event.items,
      async (ctx, item, index) =>
        ctx.step(`process-${index}`, async () => {
          if (item === "bad") throw new Error("bad item");
          return item.toUpperCase();
        }),
    );

    if (result.hasFailure) {
      const errors = result.getErrors();
      console.log(`${result.failureCount} items failed:`, errors);
    }

    const successes = result.getResults();
    console.log(`${result.successCount} items succeeded:`, successes);
  },
);

BatchResult.status is FAILED if any item failed. Call throw_if_error() to propagate the first item error as an exception, or inspect get_errors() to handle errors individually.

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)


def process_item(
    ctx: DurableContext, item: str, index: int, items: list[str]
) -> str:
    def do_process(_):
        if item == "bad":
            raise ValueError("bad item")
        return item.upper()

    return ctx.step(do_process, name=f"process-{index}")


@durable_execution
def handler(event: dict, context: DurableContext) -> None:
    result: BatchResult[str] = context.map(
        event["items"],
        process_item,
        name="process-items",
    )

    if result.has_failure:
        errors = result.get_errors()
        print(f"{result.failure_count} items failed:", errors)

    successes = result.get_results()
    print(f"{result.success_count} items succeeded:", successes)

Check result.failed() to detect item failures. Each MapError contains errorType, errorMessage, and stackTrace as plain strings. If the SDK cannot reconstruct the original exception, it throws MapIterationFailedException.

import java.util.List;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.model.MapResult;

public class MapErrorHandling extends DurableHandler<List<String>, Void> {
    @Override
    public Void handleRequest(List<String> items, DurableContext context) {
        MapResult<String> result = context.map(
                "process-items",
                items,
                String.class,
                (item, index, ctx) -> ctx.step("process-" + index, String.class, s -> {
                    if ("bad".equals(item)) throw new IllegalArgumentException("bad item");
                    return item.toUpperCase();
                }));

        var failures = result.failed();
        if (!failures.isEmpty()) {
            System.out.println(failures.size() + " items failed");
            failures.forEach(e -> System.out.println(e.errorType() + ": " + e.errorMessage()));
        }

        var successes = result.succeeded();
        System.out.println(successes.size() + " items succeeded: " + successes);
        return null;
    }
}

Map captures each item error in the result. By default (fail-fast) the first item failure completes the batch. Items still running are abandoned, and items not yet started never run. Map then returns the populated result together with a *durable.BatchError, whose Errors holds the per-item errors. Match it with errors.As and read the failures from BatchResult.Errors() and BatchResult.Failed(). Return any other error unchanged. To run every item despite failures, set a tolerance with WithCompletion.

// Package main reads item failures from a Map result. Under the default
// fail-fast policy, Map returns the populated result together with a
// *durable.BatchError.
package main

import (
    "errors"
    "fmt"
    "strings"

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

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

type Output struct {
    Succeeded int      `json:"succeeded"`
    Failed    int      `json:"failed"`
    Results   []string `json:"results"`
    Errors    []string `json:"errors"`
}

func handler(ctx durable.Context, event Input) (Output, error) {
    result, err := durable.Map(ctx, "process-items", event.Items,
        func(ctx durable.Context, item string, index int) (string, error) {
            return durable.Step(ctx, fmt.Sprintf("process-%d", index),
                func(durable.StepContext) (string, error) {
                    if item == "bad" {
                        return "", errors.New("bad item")
                    }
                    return strings.ToUpper(item), nil
                })
        })
    var batchErr *durable.BatchError
    if err != nil && !errors.As(err, &batchErr) {
        return Output{}, err
    }

    messages := make([]string, 0, result.FailureCount())
    for _, e := range result.Errors() {
        messages = append(messages, e.Error())
    }
    return Output{
        Succeeded: result.SuccessCount(),
        Failed:    result.FailureCount(),
        Results:   result.Results(),
        Errors:    messages,
    }, nil
}

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

.await returns only the results of succeeded items. It returns an error when the failure tolerance is exceeded, which the default fail-fast configuration does at the first item failure. The SDK then cancels the items still running and never starts the items not yet started. To inspect the failures, use .await_batch(), which returns the BatchResult. has_failure() is true if any item failed. Call into_result() to propagate the first item error, or inspect errors() to handle errors individually.

use aws_durable_execution_sdk as durable;
use serde::Deserialize;

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

async fn handler(event: ItemsEvent, ctx: durable::DurableContext) -> Result<(), durable::BoxError> {
    let result = ctx
        .map(event.items, |ctx, item, index| async move {
            Ok(ctx
                .step(move |_| async move {
                    if item == "bad" {
                        return Err("bad item".into());
                    }
                    Ok(item.to_uppercase())
                })
                .name(format!("process-{index}"))
                .await?)
        })
        .name("process-items")
        .await_batch()
        .await?;

    if result.has_failure() {
        let errors: Vec<_> = result.errors().collect();
        tracing::info!("{} items failed: {errors:?}", result.failure_count());
    }

    let successes: Vec<_> = result.results().collect();
    tracing::info!("{} items succeeded: {successes:?}", result.success_count());
    Ok(())
}

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

IBatchResult.HasFailure is true if any item failed. Call ThrowIfError() to propagate the first item error as an exception, or inspect GetErrors() (which returns DurableExecutionException objects) to handle errors individually.

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

public class ErrorHandlingExample
{
    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<ItemEvent, Summary>(Workflow, input, context);

    private async Task<Summary> Workflow(ItemEvent input, IDurableContext ctx)
    {
        IBatchResult<string> result = await ctx.MapAsync(
            input.Items,
            async (itemCtx, item, index, items, ct) =>
                await itemCtx.StepAsync(async (_, _) =>
                {
                    if (item == "bad") throw new InvalidOperationException("bad item");
                    return item.ToUpperInvariant();
                }, name: $"process-{index}"),
            name: "process-items");

        // BatchResult captures per-item failures rather than throwing. Inspect
        // HasFailure/GetErrors to handle them, or call ThrowIfError to propagate
        // the first failure.
        IReadOnlyList<DurableExecutionException> errors =
            result.HasFailure ? result.GetErrors() : Array.Empty<DurableExecutionException>();
        IReadOnlyList<string> successes = result.GetResults();

        return new Summary(result.SuccessCount, result.FailureCount, successes, errors);
    }
}

public record ItemEvent(IReadOnlyList<string> Items);
public record Summary(
    int Succeeded,
    int Failed,
    IReadOnlyList<string> Results,
    IReadOnlyList<DurableExecutionException> Errors);

Checkpointing¶

Checkpoint behavior depends on the nesting type. In nested mode, each item checkpoints its result in a per-item CONTEXT operation. In flat mode, the SDK omits that context checkpoint and records the item outcome with the parent map operation. Durable operations inside an item still checkpoint in both modes.

Items that have not completed when the map operation reaches its completion criteria receive no further checkpoint updates. Unless noted otherwise, the language-specific details below describe nested mode.

TypeScriptPythonJavaGoRustC#

The parent map operation also checkpoints the serialized BatchResult for observability. On replay, the SDK deserializes the BatchResult directly from that checkpoint.

For results over 256KB, the SDK cannot store the full BatchResult in the checkpoint. Instead, the SDK reconstructs the BatchResult from the checkpointed results of the individual items. In that case, the checkpoint stores a compact JSON summary, which is for observability only.

The default summary generator produces:

{
  "type": "MapResult",
  "totalCount": 5,
  "successCount": 4,
  "failureCount": 1,
  "completionReason": "ALL_COMPLETED",
  "status": "FAILED"
}

The parent map operation also checkpoints the serialized BatchResult for observability. On replay, the SDK deserializes the BatchResult directly from that checkpoint.

For results over 256KB, the SDK cannot store the full BatchResult in the checkpoint, so it re-executes the items to reconstruct it instead. In that case, the checkpoint stores the output of summary_generator, which is for observability only.

The default summary generator produces:

{
  "type": "MapResult",
  "totalCount": 5,
  "successCount": 4,
  "failureCount": 1,
  "completionReason": "ALL_COMPLETED",
  "status": "FAILED"
}

When you pass a custom MapConfig without setting summary_generator, the SDK checkpoints an empty string for large payloads.

SummaryGenerator is a callable protocol you can pass by setting summary_generator on MapConfig:

class SummaryGenerator(Protocol[T]):
    def __call__(self, result: T) -> str: ...

For results under 256KB, the SDK checkpoints the serialized MapResult payload. On replay, the SDK deserializes the MapResult directly from that checkpoint without re-executing items.

For results over 256KB, the SDK checkpoints with an empty payload and a replayChildren flag. On replay, the SDK re-executes the items to reconstruct the MapResult from their individual checkpoints.

WithNesting selects the checkpoint shape. In NestingNormal (the default) each item checkpoints its result in its own child context. In NestingFlat the SDK omits the per-item context and records the item outcome with the parent map. Durable operations inside an item still checkpoint in both modes.

The SDK checkpoints the whole BatchResult when it is at most 256KB serialized. On replay, the SDK decodes that BatchResult without running the items. A larger result is not stored. The checkpoint then keeps the child operations plus a compact record, and replay rebuilds each item from that item's own checkpoint. The record holds type, totalCount, successCount, failureCount, completionReason, status, and itemStatuses, one character per started item. Pass WithBatchSummary to add your own string under the summary key. WithBatchSerdes sets the per-item serializer and WithBatchResultSerdes the whole-result serializer.

.nesting(...) selects the checkpoint shape. In NestingMode::Normal (the default) each item checkpoints its result in its own child context. In NestingMode::Flat the SDK omits the per-item context and records the item outcome with the parent map. Durable operations inside an item still checkpoint in both modes.

The SDK checkpoints the whole batch result when it is at most 256KB serialized. On replay, the SDK reads that result without running the items. A larger result is not stored. The checkpoint then holds a compact record. In NestingMode::Normal replay rebuilds each item from that item's own checkpoint. In NestingMode::Flat replay runs the items again. The record holds type (MapResult), totalCount, successCount, failureCount, completionReason, status, and itemStatuses, one character per started item. Pass .summary(...) to add your own string under the summary key. The record is for observability only. .serdes(...) sets the per-item serializer and .result_serdes(...) the whole-result serializer.

In nested mode, the SDK reconstructs IBatchResult from the per-item child-context checkpoints without re-executing completed items. In flat mode, the SDK records item results and errors inline on the parent map operation instead.

The SDK serializes results with the ILambdaSerializer registered on ILambdaContext.Serializer; there is no per-item summary generator to configure.

Nesting map operations¶

A map function can call context.map() or context.parallel() to create nested operations. Each nested map creates its own set of child contexts.

TypeScriptPythonJavaGoRustC#

import {
  BatchResult,
  DurableContext,
  withDurableExecution,
} from "@aws/durable-execution-sdk-js";

type Region = { name: string; items: string[] };

export const handler = withDurableExecution(
  async (event: { regions: Region[] }, context: DurableContext): Promise<string[][]> => {
    const result: BatchResult<string[]> = await context.map(
      "process-regions",
      event.regions,
      async (ctx, region, index) => {
        const inner: BatchResult<string> = await ctx.map(
          `process-${region.name}`,
          region.items,
          async (innerCtx, item, i) =>
            innerCtx.step(`item-${i}`, async () => item.toUpperCase()),
        );
        return inner.getResults();
      },
    );

    return result.getResults();
  },
);

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)


def process_item(
    ctx: DurableContext, item: str, index: int, items: list[str]
) -> str:
    return ctx.step(lambda _: item.upper(), name=f"item-{index}")


def process_region(
    ctx: DurableContext, region: dict, index: int, regions: list[dict]
) -> list[str]:
    inner: BatchResult[str] = ctx.map(
        region["items"],
        process_item,
        name=f"process-{region['name']}",
    )
    return inner.get_results()


@durable_execution
def handler(event: dict, context: DurableContext) -> list[list[str]]:
    result: BatchResult[list[str]] = context.map(
        event["regions"],
        process_region,
        name="process-regions",
    )
    return result.get_results()

import java.util.List;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.TypeToken;
import software.amazon.lambda.durable.model.MapResult;

public class NestedMap extends DurableHandler<List<Region>, List<List<String>>> {
    record Region(String name, List<String> items) {}

    @Override
    public List<List<String>> handleRequest(List<Region> regions, DurableContext context) {
        MapResult<List<String>> result = context.map(
                "process-regions",
                regions,
                new TypeToken<List<String>>() {},
                (region, index, ctx) -> {
                    MapResult<String> inner = ctx.map(
                            "process-" + region.name(),
                            region.items(),
                            String.class,
                            (item, i, innerCtx) -> innerCtx.step(
                                    "item-" + i, String.class, s -> item.toUpperCase()));
                    return inner.succeeded();
                });

        return result.succeeded();
    }
}

// Package main nests a Map inside a Map. Each nested Map creates its own set
// of child contexts.
package main

import (
    "fmt"
    "strings"

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

type Region struct {
    Name  string   `json:"name"`
    Items []string `json:"items"`
}

type Input struct {
    Regions []Region `json:"regions"`
}

func handler(ctx durable.Context, event Input) ([][]string, error) {
    result, err := durable.Map(ctx, "process-regions", event.Regions,
        func(ctx durable.Context, region Region, index int) ([]string, error) {
            inner, err := durable.Map(ctx, fmt.Sprintf("process-%s", region.Name), region.Items,
                func(ctx durable.Context, item string, i int) (string, error) {
                    return durable.Step(ctx, fmt.Sprintf("item-%d", i),
                        func(durable.StepContext) (string, error) {
                            return strings.ToUpper(item), nil
                        })
                })
            if err != nil {
                return nil, err
            }
            return inner.Results(), nil
        })
    if err != nil {
        return nil, err
    }
    return result.Results(), nil
}

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

use aws_durable_execution_sdk as durable;
use serde::Deserialize;

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

#[derive(Deserialize)]
struct RegionsEvent {
    regions: Vec<Region>,
}

async fn handler(
    event: RegionsEvent,
    ctx: durable::DurableContext,
) -> Result<Vec<Vec<String>>, durable::BoxError> {
    let results = ctx
        .map(event.regions, |ctx, region, _index| async move {
            let inner = ctx
                .map(region.items, |inner_ctx, item, i| async move {
                    Ok(inner_ctx
                        .step(move |_| async move { Ok(item.to_uppercase()) })
                        .name(format!("item-{i}"))
                        .await?)
                })
                .name(format!("process-{}", region.name))
                .await?;
            Ok(inner)
        })
        .name("process-regions")
        .await?;

    Ok(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 NestedMapExample
{
    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<RegionEvent, IReadOnlyList<IReadOnlyList<string>>>(
            Workflow, input, context);

    private async Task<IReadOnlyList<IReadOnlyList<string>>> Workflow(
        RegionEvent input, IDurableContext ctx)
    {
        IBatchResult<IReadOnlyList<string>> result = await ctx.MapAsync(
            input.Regions,
            async (regionCtx, region, index, regions, ct) =>
            {
                IBatchResult<string> inner = await regionCtx.MapAsync(
                    region.Items,
                    async (itemCtx, item, i, items, innerCt) =>
                        await itemCtx.StepAsync(
                            async (_, _) => item.ToUpperInvariant(),
                            name: $"item-{i}"),
                    name: $"process-{region.Name}");
                return inner.GetResults();
            },
            name: "process-regions");

        return result.GetResults();
    }
}

public record Region(string Name, IReadOnlyList<string> Items);
public record RegionEvent(IReadOnlyList<Region> Regions);

Map¶

Apply a function to each item in a collection¶

Method signature¶

context.map¶

Map Function¶

MapConfig¶

CompletionConfig¶

Result types¶

The map function¶

Naming map operations¶

Configuration¶

Nesting¶

Completion strategies¶

Error handling¶

Checkpointing¶

Nesting map operations¶

See also¶