Configuration¶

Custom Lambda client¶

By default, the SDK initializes a Lambda client from your environment. You can provide a custom client to control the region, retry settings, credentials, or other options.

TypeScriptPythonJavaGoC#

Pass a LambdaClient instance via the config parameter of withDurableExecution.

import { LambdaClient } from "@aws-sdk/client-lambda";
import {
  DurableContext,
  withDurableExecution,
  DurableExecutionConfig,
} from "@aws/durable-execution-sdk-js";

const customClient = new LambdaClient({
  region: "us-west-2",
  maxAttempts: 5,
  retryMode: "adaptive",
});

const config: DurableExecutionConfig = { client: customClient };

export const handler = withDurableExecution(
  async (event: Record<string, unknown>, context: DurableContext) => {
    // Your durable function logic
    return { status: "success" };
  },
  config,
);

Pass a boto3 Lambda client via the boto3_client parameter of @durable_execution. The client must be a boto3 Lambda client.

import boto3
from botocore.config import Config
from aws_durable_execution_sdk_python import durable_execution, DurableContext

# Create a custom boto3 Lambda client with specific configuration
custom_lambda_client = boto3.client(
    'lambda',
    config=Config(
        retries={'max_attempts': 5, 'mode': 'adaptive'},
        connect_timeout=10,
        read_timeout=60,
    )
)

@durable_execution(boto3_client=custom_lambda_client)
def handler(event: dict, context: DurableContext) -> dict:
    # Your durable function logic
    return {"status": "success"}

Override createConfiguration() in your DurableHandler subclass and use DurableConfig.builder().withLambdaClientBuilder(...).

import java.util.Map;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.retries.AdaptiveRetryStrategy;
import software.amazon.awssdk.retries.api.RetryStrategy;
import software.amazon.awssdk.services.lambda.LambdaClient;
import software.amazon.lambda.durable.DurableConfig;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;

public class CustomLambdaClient extends DurableHandler<Map<String, Object>, Map<String, Object>> {

    @Override
    protected DurableConfig createConfiguration() {
        RetryStrategy retryStrategy = AdaptiveRetryStrategy.builder()
                .maxAttempts(5)
                .build();

        LambdaClient.Builder lambdaClientBuilder = LambdaClient.builder()
                .region(Region.US_WEST_2)
                .overrideConfiguration(o -> o.retryStrategy(retryStrategy));

        return DurableConfig.builder()
                .withLambdaClientBuilder(lambdaClientBuilder)
                .build();
    }

    @Override
    public Map<String, Object> handleRequest(Map<String, Object> event, DurableContext context) {
        // Your durable function logic
        return Map.of("status", "success");
    }
}

The SDK takes no AWS Lambda client directly. durable.WithExecutionClient takes a durable.ExecutionClient, an SDK-owned interface. Implement it over a configured *lambda.Client to control the region, retry policy, or credentials. The example wraps a client built with a custom region and retry mode, and maps each call onto the GetDurableExecutionState and CheckpointDurableExecution Lambda APIs.

The SDK's default client sets a 5 second connect timeout, a 50 second response timeout, and a 55 second total request timeout, and adds the SDK's user agent. A client you build gets none of these. Set the timeouts on its HTTP client if you need them.

The SDK builds its default client from the default AWS config. To change only the region, credentials, or retry settings of the default client, set the standard AWS environment variables or shared config files, such as AWS_REGION, AWS_MAX_ATTEMPTS, and AWS_RETRY_MODE.

package main

import (
    "context"
    "log"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    lambda "github.com/aws/aws-sdk-go-v2/service/lambda"
    "github.com/aws/aws-sdk-go-v2/service/lambda/types"

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

type Response struct {
    Status string `json:"status"`
}

func handler(ctx durable.Context, _ map[string]any) (Response, error) {
    // Your durable function logic.
    return Response{Status: "success"}, nil
}

// lambdaClientAdapter implements durable.ExecutionClient over a configured
// AWS Lambda service client. durable.WithExecutionClient takes an
// ExecutionClient, whose methods use SDK-owned request and response types,
// not AWS Lambda types. The adapter maps between the two in both
// directions, so configuring the region, retry policy, or credentials of
// the client controls the calls the SDK makes.
type lambdaClientAdapter struct {
    client *lambda.Client
}

var _ durable.ExecutionClient = (*lambdaClientAdapter)(nil)

func (a *lambdaClientAdapter) GetExecutionState(ctx context.Context, in durable.GetExecutionStateInput) (durable.GetExecutionStateOutput, error) {
    wireIn := &lambda.GetDurableExecutionStateInput{
        DurableExecutionArn: aws.String(in.ExecutionArn),
        CheckpointToken:     aws.String(in.CheckpointToken),
    }
    if in.Marker != "" {
        wireIn.Marker = aws.String(in.Marker)
    }
    wireOut, err := a.client.GetDurableExecutionState(ctx, wireIn)
    if err != nil {
        return durable.GetExecutionStateOutput{}, err
    }
    return durable.GetExecutionStateOutput{
        Operations: operationsFromWire(wireOut.Operations),
        NextMarker: aws.ToString(wireOut.NextMarker),
    }, nil
}

func (a *lambdaClientAdapter) Checkpoint(ctx context.Context, in durable.CheckpointInput) (durable.CheckpointOutput, error) {
    wireIn := &lambda.CheckpointDurableExecutionInput{
        DurableExecutionArn: aws.String(in.ExecutionArn),
        CheckpointToken:     aws.String(in.CheckpointToken),
        Updates:             updatesToWire(in.Updates),
    }
    wireOut, err := a.client.CheckpointDurableExecution(ctx, wireIn)
    if err != nil {
        return durable.CheckpointOutput{}, err
    }
    out := durable.CheckpointOutput{CheckpointToken: aws.ToString(wireOut.CheckpointToken)}
    if wireOut.NewExecutionState != nil {
        out.NewExecutionState = operationsFromWire(wireOut.NewExecutionState.Operations)
    }
    return out, nil
}

func updatesToWire(updates []durable.OperationUpdate) []types.OperationUpdate {
    if updates == nil {
        return nil
    }
    wire := make([]types.OperationUpdate, 0, len(updates))
    for _, u := range updates {
        w := types.OperationUpdate{
            Id:       u.Id,
            Type:     types.OperationType(u.Type),
            Action:   types.OperationAction(u.Action),
            SubType:  u.SubType,
            Name:     u.Name,
            ParentId: u.ParentId,
            Payload:  u.Payload,
            Error:    errorObjectToWire(u.Error),
        }
        if o := u.StepOptions; o != nil {
            w.StepOptions = &types.StepOptions{NextAttemptDelaySeconds: o.NextAttemptDelaySeconds}
        }
        if o := u.WaitOptions; o != nil {
            w.WaitOptions = &types.WaitOptions{WaitSeconds: o.WaitSeconds}
        }
        if o := u.CallbackOptions; o != nil {
            w.CallbackOptions = &types.CallbackOptions{
                TimeoutSeconds:          o.TimeoutSeconds,
                HeartbeatTimeoutSeconds: o.HeartbeatTimeoutSeconds,
            }
        }
        if o := u.ChainedInvokeOptions; o != nil {
            w.ChainedInvokeOptions = &types.ChainedInvokeOptions{
                FunctionName: o.FunctionName,
                TenantId:     o.TenantId,
            }
        }
        if o := u.ContextOptions; o != nil {
            w.ContextOptions = &types.ContextOptions{ReplayChildren: o.ReplayChildren}
        }
        wire = append(wire, w)
    }
    return wire
}

func operationsFromWire(ops []types.Operation) []durable.Operation {
    if ops == nil {
        return nil
    }
    out := make([]durable.Operation, 0, len(ops))
    for _, op := range ops {
        out = append(out, operationFromWire(op))
    }
    return out
}

func operationFromWire(op types.Operation) durable.Operation {
    o := durable.Operation{
        Id:             op.Id,
        Status:         durable.OperationStatus(op.Status),
        Type:           durable.OperationType(op.Type),
        SubType:        op.SubType,
        Name:           op.Name,
        ParentId:       op.ParentId,
        StartTimestamp: op.StartTimestamp,
        EndTimestamp:   op.EndTimestamp,
    }
    if d := op.ExecutionDetails; d != nil {
        o.ExecutionDetails = &durable.ExecutionDetails{InputPayload: d.InputPayload}
    }
    if d := op.StepDetails; d != nil {
        o.StepDetails = &durable.StepDetails{
            Attempt:              d.Attempt,
            Result:               d.Result,
            Error:                errorObjectFromWire(d.Error),
            NextAttemptTimestamp: d.NextAttemptTimestamp,
        }
    }
    if d := op.WaitDetails; d != nil {
        o.WaitDetails = &durable.WaitDetails{ScheduledEndTimestamp: d.ScheduledEndTimestamp}
    }
    if d := op.CallbackDetails; d != nil {
        o.CallbackDetails = &durable.CallbackDetails{
            CallbackId: d.CallbackId,
            Result:     d.Result,
            Error:      errorObjectFromWire(d.Error),
        }
    }
    if d := op.ChainedInvokeDetails; d != nil {
        o.ChainedInvokeDetails = &durable.ChainedInvokeDetails{
            Result: d.Result,
            Error:  errorObjectFromWire(d.Error),
        }
    }
    if d := op.ContextDetails; d != nil {
        o.ContextDetails = &durable.ContextDetails{
            Result:         d.Result,
            ReplayChildren: d.ReplayChildren,
            Error:          errorObjectFromWire(d.Error),
        }
    }
    return o
}

func errorObjectToWire(e *durable.ErrorObject) *types.ErrorObject {
    if e == nil {
        return nil
    }
    return &types.ErrorObject{
        ErrorType:    e.ErrorType,
        ErrorMessage: e.ErrorMessage,
        ErrorData:    e.ErrorData,
        StackTrace:   e.StackTrace,
    }
}

func errorObjectFromWire(e *types.ErrorObject) *durable.ErrorObject {
    if e == nil {
        return nil
    }
    return &durable.ErrorObject{
        ErrorType:    e.ErrorType,
        ErrorMessage: e.ErrorMessage,
        ErrorData:    e.ErrorData,
        StackTrace:   e.StackTrace,
    }
}

func main() {
    cfg, err := config.LoadDefaultConfig(context.Background(),
        config.WithRegion("us-west-2"),
        config.WithRetryMaxAttempts(5),
        config.WithRetryMode(aws.RetryModeAdaptive),
    )
    if err != nil {
        log.Fatal(err)
    }
    client := &lambdaClientAdapter{client: lambda.NewFromConfig(cfg)}
    durable.Start(handler, durable.WithExecutionClient(client))
}

Build an AmazonLambdaConfig (region, retry settings, and so on), construct an AmazonLambdaClient from it, and pass that IAmazonLambda as the fourth argument to DurableFunction.WrapAsync.

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

public class CustomLambdaClientExample
{
    // Create a custom Lambda client with specific region and retry configuration.
    private static readonly IAmazonLambda CustomClient = new AmazonLambdaClient(
        new AmazonLambdaConfig
        {
            RegionEndpoint = RegionEndpoint.USWest2,
            MaxErrorRetry = 5,
            RetryMode = Amazon.Runtime.RequestRetryMode.Adaptive,
        });

    // Pass the custom client as the fourth argument to DurableFunction.WrapAsync.
    public Task<DurableExecutionInvocationOutput> Handler(
        DurableExecutionInvocationInput input, ILambdaContext context)
        => DurableFunction.WrapAsync<object, Result>(Workflow, input, context, CustomClient);

    private Task<Result> Workflow(object input, IDurableContext ctx)
    {
        // Your durable function logic
        return Task.FromResult(new Result("success"));
    }

    public record Result(string Status);
}