Plugins¶

Create plugins for instrumenting durable execution lifecycle events. Each SDK calls plugin hooks at invocation, operation, and user-function boundaries, so you can integrate external tooling, such as observability platforms, without changing your handler logic. For example, you can use a plugin to emit a metric or trace span each time an operation runs. Errors thrown by a plugin are caught and logged, and never affect the execution outcome.

Experimental feature

The plugin interface is currently experimental and may change in a future release. Feedback is welcome, please share your input through our GitHub Discussion.

Define a plugin¶

A plugin implements any of the following lifecycle hooks, and the SDK calls them as the execution runs. Every hook is optional, so implement only the ones you need.

The following example plugin shows how to implement every available hook.

TypeScriptPythonJava

import {
  DurableInstrumentationPlugin,
  InvocationInfo,
  InvocationEndInfo,
  OperationInfo,
  OperationEndInfo,
  AttemptInfo,
  AttemptEndInfo,
  OperationChangeInfo,
  DurableExecutionInvocationOutput,
} from "@aws/durable-execution-sdk-js";

export const examplePlugin: DurableInstrumentationPlugin = {
  async onInvocationStart(info: InvocationInfo): Promise<void> {
    console.log(`invocation start, arn: ${info.executionArn}, first invocation: ${info.isFirstInvocation}`);
  },

  async onInvocationEnd(info: InvocationEndInfo): Promise<void> {
    console.log(`invocation end, arn: ${info.executionArn}, status: ${info.status}`);
  },

  async onOperationStart(info: OperationInfo): Promise<void> {
    console.log(`operation ${info.name} start, type: ${info.type}, replay: ${info.isReplay}`);
  },

  async onOperationEnd(info: OperationEndInfo): Promise<void> {
    console.log(`operation ${info.name} end, status: ${info.status}, error: ${info.error}`);
  },

  async onOperationAttemptStart(info: AttemptInfo): Promise<void> {
    console.log(`attempt ${info.name} start, attempt: ${info.attempt}`);
  },

  async onOperationAttemptEnd(info: AttemptEndInfo): Promise<void> {
    console.log(`attempt ${info.name} end, attempt: ${info.attempt}, outcome: ${info.outcome}`);
  },

  async wrapInvocation(
    info: InvocationInfo,
    fn: () => Promise<DurableExecutionInvocationOutput>,
  ): Promise<DurableExecutionInvocationOutput> {
    console.log(`wrap invocation, arn: ${info.executionArn}`);
    return await fn();
  },

  wrapChildContextFn(info: OperationInfo, fn: () => unknown): unknown {
    console.log(`wrap child context ${info.name}`);
    return fn();
  },

  wrapOperationAttemptFn(info: AttemptInfo, fn: () => unknown): unknown {
    console.log(`wrap attempt ${info.name}, attempt: ${info.attempt}`);
    return fn();
  },

  async onOperationChange(info: OperationChangeInfo): Promise<void> {
    console.log(`operations changed, ids: ${Object.keys(info.updatedOperations)}`);
  },

  enrichLogContext(): Record<string, string | number | boolean> {
    return { service: "orders" };
  },
};

from aws_durable_execution_sdk_python.plugin import (
    DurableInstrumentationPlugin,
    InvocationStartInfo,
    InvocationEndInfo,
    OperationStartInfo,
    OperationEndInfo,
    UserFunctionStartInfo,
    UserFunctionEndInfo,
)


class ExamplePlugin(DurableInstrumentationPlugin):
    def on_invocation_start(self, info: InvocationStartInfo) -> None:
        print(f"invocation start, arn: {info.execution_arn}, first invocation: {info.is_first_invocation}")

    def on_invocation_end(self, info: InvocationEndInfo) -> None:
        print(f"invocation end, arn: {info.execution_arn}, status: {info.status}")

    def on_operation_start(self, info: OperationStartInfo) -> None:
        print(f"operation {info.name} start, type: {info.operation_type}")

    def on_operation_end(self, info: OperationEndInfo) -> None:
        print(f"operation {info.name} end, status: {info.status}, error: {info.error}")

    def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
        print(f"user function {info.name} start, attempt: {info.attempt}")

    def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
        print(f"user function {info.name} end, attempt: {info.attempt}, outcome: {info.outcome}")

import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
import software.amazon.lambda.durable.plugin.InvocationInfo;
import software.amazon.lambda.durable.plugin.InvocationEndInfo;
import software.amazon.lambda.durable.plugin.OperationInfo;
import software.amazon.lambda.durable.plugin.OperationEndInfo;
import software.amazon.lambda.durable.plugin.UserFunctionStartInfo;
import software.amazon.lambda.durable.plugin.UserFunctionEndInfo;

public class ExamplePlugin implements DurableExecutionPlugin {
    @Override
    public void onInvocationStart(InvocationInfo info) {
        System.out.println("invocation start, arn: " + info.durableExecutionArn()
                + ", first invocation: " + info.isFirstInvocation());
    }

    @Override
    public void onInvocationEnd(InvocationEndInfo info) {
        System.out.println("invocation end, arn: " + info.durableExecutionArn()
                + ", status: " + info.invocationStatus());
    }

    @Override
    public void onOperationStart(OperationInfo info) {
        System.out.println("operation " + info.name() + " start, type: " + info.type());
    }

    @Override
    public void onOperationEnd(OperationEndInfo info) {
        System.out.println("operation " + info.name() + " end, error: " + info.error());
    }

    @Override
    public void onUserFunctionStart(UserFunctionStartInfo info) {
        System.out.println("user function " + info.name() + " start, attempt: " + info.attempt());
    }

    @Override
    public void onUserFunctionEnd(UserFunctionEndInfo info) {
        System.out.println("user function " + info.name() + " end, attempt: " + info.attempt()
                + ", succeeded: " + info.succeeded());
    }
}

The hooks fire at three boundaries: invocation, operation, and the user function. Each hook receives an info object describing that event, including fields like operation identity, timestamps, or error details.

Invocation hooks¶

Invocation hooks run when a Lambda invocation of the execution starts or ends. The SDK passes an isFirstInvocation flag to the start hook, which indicates whether this is the execution's first invocation or a replay.

TypeScriptPythonJava

async onInvocationStart(info: InvocationInfo): Promise<void> {
  console.log(`invocation start, arn: ${info.executionArn}, first invocation: ${info.isFirstInvocation}`);
},

async onInvocationEnd(info: InvocationEndInfo): Promise<void> {
  console.log(`invocation end, arn: ${info.executionArn}, status: ${info.status}`);
},

def on_invocation_start(self, info: InvocationStartInfo) -> None:
    print(f"invocation start, arn: {info.execution_arn}, first invocation: {info.is_first_invocation}")

def on_invocation_end(self, info: InvocationEndInfo) -> None:
    print(f"invocation end, arn: {info.execution_arn}, status: {info.status}")

@Override
public void onInvocationStart(InvocationInfo info) {
    System.out.println("invocation start, arn: " + info.durableExecutionArn()
            + ", first invocation: " + info.isFirstInvocation());
}

@Override
public void onInvocationEnd(InvocationEndInfo info) {
    System.out.println("invocation end, arn: " + info.durableExecutionArn()
            + ", status: " + info.invocationStatus());
}

Operation hooks¶

Operation hooks run when a durable operation, such as a step or wait, starts or reaches a terminal status.

TypeScriptPythonJava

async onOperationStart(info: OperationInfo): Promise<void> {
  console.log(`operation ${info.name} start, type: ${info.type}, replay: ${info.isReplay}`);
},

async onOperationEnd(info: OperationEndInfo): Promise<void> {
  console.log(`operation ${info.name} end, status: ${info.status}, error: ${info.error}`);
},

The isReplay field on OperationInfo indicates whether the hook is firing for the first time or during a replay.

def on_operation_start(self, info: OperationStartInfo) -> None:
    print(f"operation {info.name} start, type: {info.operation_type}")

def on_operation_end(self, info: OperationEndInfo) -> None:
    print(f"operation {info.name} end, status: {info.status}, error: {info.error}")

@Override
public void onOperationStart(OperationInfo info) {
    System.out.println("operation " + info.name() + " start, type: " + info.type());
}

@Override
public void onOperationEnd(OperationEndInfo info) {
    System.out.println("operation " + info.name() + " end, error: " + info.error());
}

User-function hooks¶

User-function hooks run when the code you supply for an operation begins or finishes, on that operation's thread.

TypeScriptPythonJava

async onOperationAttemptStart(info: AttemptInfo): Promise<void> {
  console.log(`attempt ${info.name} start, attempt: ${info.attempt}`);
},

async onOperationAttemptEnd(info: AttemptEndInfo): Promise<void> {
  console.log(`attempt ${info.name} end, attempt: ${info.attempt}, outcome: ${info.outcome}`);
},

These hooks fire around each step or wait-for-condition attempt. Child context functions are instrumented through wrapChildContextFn, a TypeScript-only hook described in TypeScript-only hooks.

def on_user_function_start(self, info: UserFunctionStartInfo) -> None:
    print(f"user function {info.name} start, attempt: {info.attempt}")

def on_user_function_end(self, info: UserFunctionEndInfo) -> None:
    print(f"user function {info.name} end, attempt: {info.attempt}, outcome: {info.outcome}")

These hooks fire around step attempts, child context functions, and wait-for-condition checks.

@Override
public void onUserFunctionStart(UserFunctionStartInfo info) {
    System.out.println("user function " + info.name() + " start, attempt: " + info.attempt());
}

@Override
public void onUserFunctionEnd(UserFunctionEndInfo info) {
    System.out.println("user function " + info.name() + " end, attempt: " + info.attempt()
            + ", succeeded: " + info.succeeded());
}

These hooks fire around step attempts, child context functions, and wait-for-condition checks.

Checkpoint-change hook¶

The checkpoint-change hook runs when a checkpoint response reports that operations changed status. It receives the operations that changed and a snapshot of all operations.

TypeScriptPythonJava

async onOperationChange(info: OperationChangeInfo): Promise<void> {
  console.log(`operations changed, ids: ${Object.keys(info.updatedOperations)}`);
},

This hook is a work in progress and is not yet available in the Python SDK.

This hook is a work in progress and is not yet available in the Java SDK.

Use a plugin¶

Provide plugins in the durable handler configuration. The SDK calls each plugin in the order you provide.

TypeScriptPythonJava

import { DurableContext, withDurableExecution } from "@aws/durable-execution-sdk-js";
import { examplePlugin } from "./example-plugin";

export const handler = withDurableExecution(
  async (event, context: DurableContext) => {
    return await context.step("process", async () => "done");
  },
  { plugins: [examplePlugin] },
);

from aws_durable_execution_sdk_python import DurableContext, durable_execution

from example_plugin import ExamplePlugin


@durable_execution(plugins=[ExamplePlugin()])
def handler(event: dict, context: DurableContext):
    return context.step(lambda ctx: "done", name="process")

import software.amazon.lambda.durable.DurableConfig;
import software.amazon.lambda.durable.DurableContext;
import software.amazon.lambda.durable.DurableHandler;

public class ExampleHandler extends DurableHandler<Object, String> {
    @Override
    protected DurableConfig createConfiguration() {
        return DurableConfig.builder()
                .withPlugins(new ExamplePlugin())
                .build();
    }

    @Override
    protected String handleRequest(Object event, DurableContext context) {
        return context.step("process", String.class, stepCtx -> "done");
    }
}

Logging from a plugin¶

A plugin can add custom fields to log entries so that logs emitted during the execution carry extra context.

TypeScriptPythonJava

enrichLogContext is called for each log entry, and the fields it returns are added to that entry. The enrichment applies wherever the SDK logs. See Logging for more about the SDK logger.

enrichLogContext(): Record<string, string | number | boolean> {
  return { service: "orders" };
},

A plugin can enrich logs by installing a standard logging.Filter on the root logger, usually in its constructor. The filter runs for every log record on the emitting thread, so it can add fields that reflect the plugin's current state. See Logging for more about the SDK logger.

import logging

from aws_durable_execution_sdk_python.plugin import DurableInstrumentationPlugin


class ServiceLogFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.service = "orders"
        return True


class ExamplePlugin(DurableInstrumentationPlugin):
    def __init__(self) -> None:
        for handler in logging.getLogger().handlers:
            handler.addFilter(ServiceLogFilter())

The onUserFunctionStart and onUserFunctionEnd hooks run on the same thread as the user function, so a plugin can add fields to the SLF4J MDC in those hooks and the logs the function emits include those fields. See Logging for more information.

@Override
public void onUserFunctionStart(UserFunctionStartInfo info) {
    MDC.put("operationName", info.name());
}

@Override
public void onUserFunctionEnd(UserFunctionEndInfo info) {
    MDC.remove("operationName");
}

TypeScript-only hooks¶

TypeScript only

These hooks exist only in the TypeScript SDK and will not be added to the Python or Java SDKs.

JavaScript has no thread-local storage to share context between separate start and end hooks. The TypeScript SDK solves this with hooks that each receive a unit of work as a function and call it, a common JavaScript callback pattern, so a plugin can keep context active from the start to the end of the work.

The Python and Java SDKs do not need these hooks. Their start and end hooks run on the same thread as the work, so a plugin can establish context in the start hook and tear it down in the end hook.

Wrap invocation¶

This hook surrounds the entire handler and every operation it runs.

async wrapInvocation(
  info: InvocationInfo,
  fn: () => Promise<DurableExecutionInvocationOutput>,
): Promise<DurableExecutionInvocationOutput> {
  console.log(`wrap invocation, arn: ${info.executionArn}`);
  return await fn();
},

Wrap child context¶

This hook surrounds a single child context function.

wrapChildContextFn(info: OperationInfo, fn: () => unknown): unknown {
  console.log(`wrap child context ${info.name}`);
  return fn();
},

Wrap operation attempt¶

This hook surrounds a single step or wait-for-condition attempt, running again for each retry.

wrapOperationAttemptFn(info: AttemptInfo, fn: () => unknown): unknown {
  console.log(`wrap attempt ${info.name}, attempt: ${info.attempt}`);
  return fn();
},