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()

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().

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.

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();
        }
    }
}

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();
        }
    }
}

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

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),
    };
  },
);

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

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


def task_1(ctx: DurableContext) -> str:
    return ctx.step(lambda _: "ok", name="task-1")


def task_2(ctx: DurableContext) -> str:
    def fail(_):
        raise ValueError("task 2 failed")

    return ctx.step(fail, name="task-2")


def task_3(ctx: DurableContext) -> str:
    return ctx.step(lambda _: "ok", name="task-3")


@durable_execution
def handler(event: dict, context: DurableContext) -> dict:
    result: BatchResult[str] = context.parallel(
        [task_1, task_2, task_3],
        name="tasks",
        config=ParallelConfig(
            completion_config=CompletionConfig(tolerated_failure_count=1)
        ),
    )
    return {
        "succeeded": result.success_count,
        "failed": result.failure_count,
        "results": result.get_results(),
        "errors": [e.message for e in result.get_errors()],
    }

Check result.failed() > 0 (where result is a ParallelResult) to detect branch failures. To propagate a branch error, call .get() on the DurableFuture<T> for that branch to rethrow the original exception. This will throw ParallelBranchFailedException if the SDK cannot reconstruct the original.

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.ParallelConfig;
import software.amazon.lambda.durable.model.ParallelResult;

public class ErrorHandling extends DurableHandler<Void, ParallelResult> {
    @Override
    public ParallelResult handleRequest(Void input, DurableContext context) {
        var config = ParallelConfig.builder()
                .completionConfig(CompletionConfig.toleratedFailureCount(1))
                .build();

        try (var parallel = context.parallel("tasks", config)) {
            parallel.branch("task-1", String.class,
                    ctx -> ctx.step("task-1", String.class, s -> "ok"));
            parallel.branch("task-2", String.class, ctx -> ctx.step("task-2", String.class, s -> {
                throw new RuntimeException("task 2 failed");
            }));
            parallel.branch("task-3", String.class,
                    ctx -> ctx.step("task-3", String.class, s -> "ok"));
            return parallel.get();
        }
    }
}

The parent parallel 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 branches. In that case, the checkpoint stores a compact JSON summary, which is for observability only.

The default summary generator produces:

{
  "type": "ParallelResult",
  "totalCount": 3,
  "successCount": 2,
  "failureCount": 1,
  "startedCount": 0,
  "completionReason": "ALL_COMPLETED",
  "status": "FAILED"
}

The parent parallel 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 branches 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": "ParallelResult",
  "totalCount": 3,
  "successCount": 2,
  "failureCount": 1,
  "startedCount": 0,
  "completionReason": "ALL_COMPLETED",
  "status": "FAILED"
}

When you pass a custom ParallelConfig 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 ParallelConfig:

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

The parent parallel operation checkpoints no result payload. On replay, the SDK always re-executes the branches to reconstruct the ParallelResult from their individual checkpoints.