Parallel¶

// Named overload
parallel<TOutput>(
  name: string | undefined,
  branches: (ParallelFunc<TOutput> | NamedParallelBranch<TOutput>)[],
  config?: ParallelConfig<TOutput>,
): DurablePromise<BatchResult<TOutput>>

// Unnamed overload
parallel<TOutput>(
  branches: (ParallelFunc<TOutput> | NamedParallelBranch<TOutput>)[],
  config?: ParallelConfig<TOutput>,
): DurablePromise<BatchResult<TOutput>>

Parameters:

• name (optional) A name for the parallel operation. Pass undefined to omit.
• branches An array of ParallelFunc or NamedParallelBranch objects.
• config (optional) A ParallelConfig<TOutput> object.

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

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

ParallelFunc / NamedParallelBranch

Each element in branches is either a plain function or a named branch object:

type ParallelFunc<TResult> = (context: DurableContext) => Promise<TResult>

interface NamedParallelBranch<TResult> {
  name?: string;
  func: ParallelFunc<TResult>;
}

• name (optional) A name for this branch.
• func An async function receiving a DurableContext and returning Promise<TResult>.

Use NamedParallelBranch to give an inline lambda a name without defining a named function.

def parallel(
    self,
    functions: Sequence[Callable[[DurableContext], T]],
    name: str | None = None,
    config: ParallelConfig | None = None,
) -> BatchResult[T]: ...

Parameters:

• functions A sequence of callables, each receiving a DurableContext and returning T.
• name (optional) A name for the parallel operation.
• config (optional) A ParallelConfig object.

Returns: BatchResult[T].

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

Each element in functions is a plain callable (ctx: DurableContext) -> T. Python has no named-branch wrapper type.

ParallelDurableFuture parallel(String name)
ParallelDurableFuture parallel(String name, ParallelConfig config)

Parameters:

• name (required) A name for the parallel operation.
• config (optional) A ParallelConfig object.

Returns: ParallelResult from get().

Throws: Branch exceptions are captured in ParallelResult. Inspect succeeded and failed counts to detect failures.

ParallelDurableFuture

parallel() returns a ParallelDurableFuture. Call branch() to register and immediately start each branch, then call get() to block until all complete.

interface ParallelDurableFuture extends AutoCloseable, DurableFuture<ParallelResult> {
    <T> DurableFuture<T> branch(String name, Class<T> resultType,
                                Function<DurableContext, T> func);
    <T> DurableFuture<T> branch(String name, TypeToken<T> resultType,
                                Function<DurableContext, T> func);
    <T> DurableFuture<T> branch(String name, Class<T> resultType,
                                Function<DurableContext, T> func,
                                ParallelBranchConfig config);
    ParallelResult get();   // blocks until all branches complete
    void close();           // calls get() if not already called
}

Parameters for branch():

• name (required) A name for this branch.
• resultType Class<T> or TypeToken<T> for deserialization.
• func Function<DurableContext, T> to execute in the branch's child context.
• config (optional) ParallelBranchConfig for per-branch serialization.

Returns from branch(): DurableFuture<T>. DurableFuture<T> is the common return type for all async Java SDK operations. Call .get() on it after parallel.get() returns to retrieve that branch's individual result.

ParallelBranchConfig

ParallelBranchConfig sets a custom SerDes for a single branch, overriding the handler-level default set on DurableConfig.

ParallelBranchConfig.builder()
    .serDes(SerDes)  // optional
    .build()

Task<IBatchResult<T>> ParallelAsync<T>(
    IReadOnlyList<Func<IDurableContext, CancellationToken, Task<T>>> branches,
    string? name = null,
    ParallelConfig? config = null,
    CancellationToken cancellationToken = default)

Task<IBatchResult<T>> ParallelAsync<T>(
    IReadOnlyList<DurableBranch<T>> branches,
    string? name = null,
    ParallelConfig? config = null,
    CancellationToken cancellationToken = default)

Parameters:

• branches The branches to run concurrently, either as plain Func<IDurableContext, CancellationToken, Task<T>> delegates or as named DurableBranch<T> records.
• name (optional) A name for the parallel operation. Omit it to infer one from the call site.
• config (optional) A ParallelConfig object.
• cancellationToken (optional) A token linked with the SDK's workflow-shutdown signal, forwarded to each branch.

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

Throws: Branch exceptions are captured in the IBatchResult. A completion-criteria violation surfaces as ParallelException when awaited. Call ThrowIfError() to re-throw the first branch failure explicitly.

DurableBranch<T>

Use the second overload to give each branch a name. DurableBranch<T> is a record pairing a name with the branch function; the name surfaces on IBatchItem<T>.Name.

public sealed record DurableBranch<T>(
    string Name,
    Func<IDurableContext, CancellationToken, Task<T>> Func);

• Name (required) A name for this branch.
• Func An async function receiving the branch's IDurableContext and a CancellationToken, returning Task<T>.

interface ParallelConfig<TResult> {
  maxConcurrency?: number;
  completionConfig?: CompletionConfig;
  serdes?: Serdes<BatchResult<TResult>>;
  itemSerdes?: Serdes<TResult>;
  nesting?: NestingType;
}

Parameters:

• maxConcurrency (optional) Maximum branches running at once. Default: unlimited.
• completionConfig (optional) When to stop. Default: wait for all branches.
• serdes (optional) Custom Serdes for the BatchResult.
• itemSerdes (optional) Custom Serdes for individual branch results.
• nesting (optional) NestingType.NESTED (default) or NestingType.FLAT. FLAT reduces operation overhead by ~30% at the cost of lower observability.

@dataclass(frozen=True)
class ParallelConfig:
    max_concurrency: int | None = None
    completion_config: CompletionConfig = CompletionConfig.all_successful()
    serdes: SerDes | None = None
    item_serdes: SerDes | None = None
    summary_generator: SummaryGenerator | None = None

Parameters:

• max_concurrency (optional) Maximum branches running at once. Default: unlimited.
• completion_config (optional) When to stop. Default: CompletionConfig.all_successful().
• serdes (optional) Custom SerDes for the BatchResult.
• item_serdes (optional) Custom SerDes for individual branch results.
• summary_generator (optional) A callable invoked when the serialized BatchResult exceeds 256KB. See Checkpointing.

ParallelConfig.builder()
    .maxConcurrency(Integer)      // optional
    .completionConfig(CompletionConfig)  // optional
    .build()

Parameters:

• maxConcurrency (optional) Maximum branches running at once. Default: unlimited.
• completionConfig (optional) When to stop. Default: CompletionConfig.allCompleted().

public sealed class ParallelConfig
{
    public int? MaxConcurrency { get; set; }              // null = unlimited
    public CompletionConfig CompletionConfig { get; set; } // default AllSuccessful()
    public NestingType NestingType { get; set; }          // default Nested
}

Parameters:

• MaxConcurrency (optional) Maximum branches running at once. null (default) = unlimited. Must be at least 1 when set.
• CompletionConfig (optional) When to stop. Default: CompletionConfig.AllSuccessful().
• NestingType (optional) NestingType.Nested (default) or NestingType.Flat. Flat records per-branch results inline on the parallel operation instead of emitting a per-branch CONTEXT checkpoint.

The IBatchResult is reconstructed from per-branch checkpoints, which are serialized with the ILambdaSerializer registered on ILambdaContext.Serializer; there is no per-operation serializer slot.

CompletionConfig.AllSuccessful()  // tolerate zero failures (the default)
CompletionConfig.AllCompleted()   // run every branch; never auto-throws
CompletionConfig.FirstSuccessful() // resolve once one branch succeeds

// Or set the individual properties directly:
new CompletionConfig { MinSuccessful = 2 }
new CompletionConfig { ToleratedFailureCount = 1 }
new CompletionConfig { ToleratedFailurePercentage = 0.25 } // ratio in [0.0, 1.0]

AllSuccessful() is equivalent to ToleratedFailureCount = 0, and FirstSuccessful() is equivalent to MinSuccessful = 1. Multiple criteria combine: the operation resolves as soon as any criterion is met or violated.

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 branch, in input order. Iterate with item.index for branch-indexed access when some branches fail.
• getResults() results of succeeded branches, preserving input order
• getErrors() ChildContextError[] for failed branches
• succeeded() / failed() / started() BatchItem[] filtered by status
• successCount / failureCount / startedCount / totalCount branch counts
• status SUCCEEDED if no failures, FAILED otherwise
• completionReason why the operation completed. See Completion strategies.
• hasFailure true if any branch failed
• throwIfError() throws the first branch error, if any

type CompletionReason =
  | "ALL_COMPLETED"
  | "MIN_SUCCESSFUL_REACHED"
  | "FAILURE_TOLERANCE_EXCEEDED"

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

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

• index position of this branch in the input array
• status SUCCEEDED, FAILED, or STARTED (not yet complete)
• result the branch return value, present when status is SUCCEEDED
• error the captured error, present when status is FAILED

@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: ...

class CompletionReason(Enum):
    ALL_COMPLETED             = "ALL_COMPLETED"
    MIN_SUCCESSFUL_REACHED    = "MIN_SUCCESSFUL_REACHED"
    FAILURE_TOLERANCE_EXCEEDED = "FAILURE_TOLERANCE_EXCEEDED"

• all all BatchItem entries, one per branch, in input order. Iterate with item.index for branch-indexed access when some branches fail.
• get_results() results of succeeded branches, preserving input order
• get_errors() list[ErrorObject] for failed branches
• succeeded() / failed() / started() BatchItem lists filtered by status
• success_count / failure_count / started_count / total_count branch counts
• status BatchItemStatus.SUCCEEDED if no failures, FAILED otherwise
• completion_reason why the operation completed. See Completion strategies.
• has_failure True if any branch failed
• throw_if_error() raises the first branch error as a CallableRuntimeError
• to_dict() serializes to a plain dict. Serializability depends on R.

@dataclass(frozen=True)
class BatchItem(Generic[R]):
    index: int
    status: BatchItemStatus
    result: R | None = None
    error: ErrorObject | None = None

    def to_dict(self) -> dict: ...

class BatchItemStatus(Enum):
    SUCCEEDED = "SUCCEEDED"
    FAILED    = "FAILED"
    STARTED   = "STARTED"

• index position of this branch in the input sequence
• status SUCCEEDED, FAILED, or STARTED (not yet complete)
• result the branch return value, present when status is SUCCEEDED
• error ErrorObject with the captured error, present when status is FAILED
• to_dict() serializes to a plain dict. Serializability of result depends on R.

record ParallelResult(
    int size,
    int succeeded,
    int failed,
    ConcurrencyCompletionStatus completionStatus
) {}

enum ConcurrencyCompletionStatus {
    ALL_COMPLETED,
    MIN_SUCCESSFUL_REACHED,
    FAILURE_TOLERANCE_EXCEEDED
}

• size total number of registered branches
• succeeded number of branches that succeeded
• failed number of branches that failed
• completionStatus why the operation completed. See Completion strategies.

ConcurrencyCompletionStatus.isSucceeded() returns true for both ALL_COMPLETED and MIN_SUCCESSFUL_REACHED. To check if any branch failed, use result.failed() > 0 (where result is a ParallelResult).

ParallelResult contains only aggregate counts. To get individual branch results, hold the DurableFuture<T> returned by each branch() call and call .get() on it after parallel.get() returns. Results are available in the order branches were registered.

public interface IBatchResult<T> : IBatchResult
{
    IReadOnlyList<IBatchItem<T>> All { get; }
    IReadOnlyList<IBatchItem<T>> Succeeded { get; }
    IReadOnlyList<IBatchItem<T>> Failed { get; }
    IReadOnlyList<IBatchItem<T>> Started { get; }
    IReadOnlyList<T> GetResults();
    IReadOnlyList<DurableExecutionException> GetErrors();
    void ThrowIfError();
}

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 branch, in original index order. Iterate with item.Index for branch-indexed access when some branches fail.
• Succeeded / Failed / Started IBatchItem lists filtered by status, in original index order
• GetResults() results of succeeded branches, preserving input order. Skips failed and started items, so it never throws on partial-failure batches
• GetErrors() DurableExecutionException list for failed branches
• SuccessCount / FailureCount / StartedCount / TotalCount branch counts
• HasFailure true if any branch failed
• CompletionReason why the operation completed. See Completion strategies.
• ThrowIfError() throws the first failed branch's Error, if any

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

public enum BatchItemStatus
{
    Succeeded,
    Failed,
    Started
}

public enum CompletionReason
{
    AllCompleted,
    MinSuccessfulReached,
    FailureToleranceExceeded
}

• Index zero-based position of this branch in the input list
• Name the branch name (from DurableBranch<T>.Name), if any
• Status Succeeded, Failed, or Started (not dispatched before the operation resolved)
• Result the branch return value, present when Status is Succeeded
• Error the captured error, present when Status is Failed

A branch is a ParallelFunc (plain async function) or a NamedParallelBranch (object with name and func). Use NamedParallelBranch to give an inline lambda a name without defining a named function.

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

// Plain function branch
async function taskA(ctx: DurableContext): Promise<string> {
  return ctx.step("run-a", async () => "a done");
}

export const handler = withDurableExecution(
  async (event: any, context: DurableContext): Promise<string[]> => {
    const result: BatchResult<string> = await context.parallel("process", [
      // ParallelFunc: plain async function
      taskA,
      // NamedParallelBranch: object with name and func
      { name: "task-b", func: async (ctx) => ctx.step("run-b", async () => "b done") },
    ]);

    return result.getResults();
  },
);

Branch functions are synchronous callables that receive a DurableContext and return T.

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)


def task_a(ctx: DurableContext) -> str:
    return ctx.step(lambda _: "a done", name="run-a")


def task_b(ctx: DurableContext) -> str:
    return ctx.step(lambda _: "b done", name="run-b")


@durable_execution
def handler(event: dict, context: DurableContext) -> list[str]:
    result: BatchResult[str] = context.parallel(
        [task_a, task_b],
        name="process",
    )
    return result.to_dict()

Each branch is registered via ParallelDurableFuture.branch(). The branch function is a synchronous Function<DurableContext, T>.

import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;
import software.amazon.lambda.durable.model.ParallelResult;

public class NamedBranches extends DurableHandler<Void, ParallelResult> {
    @Override
    public ParallelResult handleRequest(Void input, DurableContext context) {
        try (var parallel = context.parallel("process")) {
            parallel.branch("task-a", String.class,
                    ctx -> ctx.step("run-a", String.class, s -> "a done"));
            parallel.branch("task-b", String.class,
                    ctx -> ctx.step("run-b", String.class, s -> "b done"));
            return parallel.get();
        }
    }
}

A branch is a plain Func<IDurableContext, CancellationToken, Task<T>>, or a DurableBranch<T> record to give it a name. Use the named overload to surface a name on IBatchItem<T>.Name without defining a named method.

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

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

    private async Task<IReadOnlyList<string>> Workflow(object input, IDurableContext ctx)
    {
        // Each DurableBranch carries a name surfaced on IBatchItem.Name.
        IBatchResult<string> result = await ctx.ParallelAsync(
            new[]
            {
                new DurableBranch<string>("task-a", async (branch, ct) =>
                    await branch.StepAsync(async (_, _) => "a done", name: "run-a")),
                new DurableBranch<string>("task-b", async (branch, ct) =>
                    await branch.StepAsync(async (_, _) => "b done", name: "run-b")),
            },
            name: "process");

        return result.GetResults();
    }
}

Capture arguments in a closure:

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

export const handler = withDurableExecution(
  async (event: any, context: DurableContext): Promise<string[]> => {
    const items = ["a", "b", "c"];

    const result: BatchResult<string> = await context.parallel(
      "process-items",
      items.map((item) => async (ctx: DurableContext) =>
        ctx.step(`process-${item}`, async () => `processed ${item}`),
      ),
    );

    return result.getResults();
  },
);

Use a factory function to bind arguments. Avoid using loop variables directly in lambdas, as Python closures capture by reference.

from aws_durable_execution_sdk_python import (
    BatchResult,
    DurableContext,
    durable_execution,
)


def make_branch(item: str):
    def branch(ctx: DurableContext) -> str:
        return ctx.step(lambda _: f"processed {item}", name=f"process-{item}")

    return branch


@durable_execution
def handler(event: dict, context: DurableContext) -> list[str]:
    items = ["a", "b", "c"]
    result: BatchResult[str] = context.parallel(
        [make_branch(item) for item in items],
        name="process-items",
    )
    return result.to_dict()

Capture arguments in a lambda. Java lambdas require effectively final variables.

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

public class PassArguments extends DurableHandler<Void, ParallelResult> {
    @Override
    public ParallelResult handleRequest(Void input, DurableContext context) {
        var items = List.of("a", "b", "c");

        try (var parallel = context.parallel("process-items")) {
            for (var item : items) {
                parallel.branch("process-" + item, String.class,
                        ctx -> ctx.step("run-" + item, String.class, s -> "processed " + item));
            }
            return parallel.get();
        }
    }
}

Capture arguments in the closure. Copy the loop variable to a local so each branch captures its own value.

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

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

    private async Task<IReadOnlyList<string>> Workflow(object input, IDurableContext ctx)
    {
        var items = new[] { "a", "b", "c" };

        // Capture each item in the closure. Copy the loop variable to a local
        // so every branch captures its own value.
        var branches = new List<DurableBranch<string>>();
        foreach (var item in items)
        {
            var captured = item;
            branches.Add(new DurableBranch<string>($"process-{captured}",
                async (branch, ct) => await branch.StepAsync(
                    async (_, _) => $"processed {captured}", name: $"run-{captured}")));
        }

        IBatchResult<string> result = await ctx.ParallelAsync(branches, name: "process-items");

        return result.GetResults();
    }
}

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

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

The name is always required. Each branch() call also requires a name. Pass null to omit it.

The name is the optional name argument. Omit it to infer one from the call site. Use the DurableBranch<T> overload to name each branch.

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

The BatchResult's completion_reason indicates the stop condition with which the parallel operation completed. Branches that were never started appear in result.all with status STARTED.

The ParallelResult's completionStatus indicates the stop condition with which the parallel operation completed. All registered branches (including those never started) are counted in size.

The IBatchResult's CompletionReason indicates the stop condition with which the parallel operation completed. Branches that were not dispatched before the operation resolved appear in result.All with status Started.

When the operation resolves with FailureToleranceExceeded, awaiting it throws ParallelException, which carries the aggregate result on Result.

BatchResult.status is FAILED if any branch failed. Call throwIfError() to propagate the first branch 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: any, context: DurableContext) => {
    const result: BatchResult<string> = await context.parallel(
      "tasks",
      [
        async (ctx) => ctx.step("task-1", async () => "ok"),
        async (ctx) =>
          ctx.step("task-2", async () => {
            throw new Error("task 2 failed");
          }),
        async (ctx) => ctx.step("task-3", async () => "ok"),
      ],
      { completionConfig: { toleratedFailureCount: 1 } },
    );

    return {
      succeeded: result.successCount,
      failed: result.failureCount,
      results: result.getResults(),
      errors: result.getErrors().map((e) => e.message),
    };
  },
);