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.

Go support

The Go SDK has a plugin API, and it is experimental. For the reference, see the Go package documentation for durable.Plugin and durable.WithPlugins.

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.

TypeScriptPythonJavaRust

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 onOperationChange(info: OperationChangeInfo): Promise<void> {
    console.log(`operations changed, ids: ${Object.keys(info.updatedOperations)}`);
  },

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

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

from aws_durable_execution_sdk_python.plugin import (
    DurableInstrumentationPlugin,
    InvocationStartInfo,
    InvocationEndInfo,
    OperationStartInfo,
    OperationEndInfo,
    OperationChangeInfo,
    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_operation_change(self, info: OperationChangeInfo) -> None:
        print("operations changed", "ids:", list(info.updated_operations.keys()))

    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.OperationChangeInfo;
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 onOperationChange(OperationChangeInfo info) {
        System.out.println("operations changed, ids: " + info.updatedOperations().keySet());
    }

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

A plugin implements durable::plugin::Plugin, and a PluginFactory returns zero or one plugin for each invocation. Every hook has a default implementation, so implement only the hooks you need. The operation, attempt, change, and user-function hooks are synchronous and return HookWork, which is Ok(None) or a future that does further work. The invocation hooks are asynchronous and return a HookFuture. An Err or a panic from a hook is logged and changes nothing else. A binary built with panic = "abort" aborts at the panic instead. The plugin's wrap_invocation enters a tracing span with a service field, which Logging from a plugin describes.

use aws_durable_execution_sdk as durable;
use durable::plugin::{
    AttemptEndInfo, AttemptInfo, HookFuture, HookWork, InvocationEndInfo, InvocationInfo,
    OperationChangeInfo, OperationInfo, Plugin, PluginFactory, PropagationInput,
};
use durable::{OperationKind, OperationStatus, WireError};

struct ExamplePlugin;

impl Plugin for ExamplePlugin {
    fn on_invocation_start<'a>(&'a self, info: &'a InvocationInfo) -> HookFuture<'a> {
        Box::pin(async move {
            tracing::info!(
                "invocation start, arn: {}, first invocation: {}",
                info.execution_arn(),
                info.is_first_invocation()
            );
            Ok(())
        })
    }

    fn on_invocation_end<'a>(&'a self, info: &'a InvocationEndInfo) -> HookFuture<'a> {
        Box::pin(async move {
            tracing::info!(
                "invocation end, arn: {}, status: {}",
                info.execution_arn(),
                info.status().wire_str()
            );
            Ok(())
        })
    }

    fn on_operation_start(&self, info: &OperationInfo) -> HookWork {
        tracing::info!(
            "operation {} start, type: {:?}, replay: {}",
            info.name().unwrap_or_default(),
            info.kind().map(OperationKind::wire_str),
            info.is_replay()
        );
        Ok(None)
    }

    fn on_operation_end(&self, info: &OperationInfo) -> HookWork {
        tracing::info!(
            "operation {} end, status: {:?}, error: {:?}",
            info.name().unwrap_or_default(),
            info.status().map(OperationStatus::wire_str),
            info.error().map(ToString::to_string)
        );
        Ok(None)
    }

    fn on_operation_change(&self, info: &OperationChangeInfo) -> HookWork {
        let ids: Vec<&str> = info
            .updated_operations()
            .iter()
            .map(OperationInfo::id)
            .collect();
        tracing::info!("operations changed, ids: {ids:?}");
        Ok(None)
    }

    fn on_attempt_start(&self, info: &AttemptInfo) -> HookWork {
        tracing::info!(
            "attempt {} start, attempt: {}",
            info.operation().name().unwrap_or_default(),
            info.attempt()
        );
        Ok(None)
    }

    fn on_attempt_end(&self, info: &AttemptEndInfo) -> HookWork {
        tracing::info!(
            "attempt {} end, attempt: {}, outcome: {}",
            info.operation().name().unwrap_or_default(),
            info.attempt(),
            info.outcome().wire_str()
        );
        Ok(None)
    }

    fn on_user_function_start(&self, info: &OperationInfo) -> HookWork {
        tracing::info!(
            "user function {} start, attempt: {:?}",
            info.name().unwrap_or_default(),
            info.attempt()
        );
        Ok(None)
    }

    fn on_user_function_end(
        &self,
        info: &OperationInfo,
        outcome: Result<(), &WireError>,
    ) -> HookWork {
        tracing::info!(
            "user function {} end, attempt: {:?}, outcome: {:?}",
            info.name().unwrap_or_default(),
            info.attempt(),
            outcome.map_err(ToString::to_string)
        );
        Ok(None)
    }

    fn propagation_trace_id(
        &self,
        _input: &PropagationInput,
    ) -> Result<Option<String>, durable::BoxError> {
        Ok(None)
    }

    fn wrap_invocation(&self, info: &InvocationInfo, next: &mut dyn FnMut()) {
        tracing::info!("wrap invocation, arn: {}", info.execution_arn());
        tracing::info_span!("example_plugin", service = "orders").in_scope(next);
    }

    fn wrap_user_function(&self, info: &OperationInfo, next: &mut dyn FnMut()) {
        tracing::info!(
            "wrap user function {}, attempt: {:?}",
            info.name().unwrap_or_default(),
            info.attempt()
        );
        next();
    }
}

struct ExamplePluginFactory;

impl PluginFactory for ExamplePluginFactory {
    fn create_plugin(
        &self,
        _info: &InvocationInfo,
    ) -> Result<Option<Box<dyn Plugin>>, durable::BoxError> {
        Ok(Some(Box::new(ExamplePlugin)))
    }
}

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.

TypeScriptPythonJavaRust

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

impl Plugin for ExamplePlugin {
    fn on_invocation_start<'a>(&'a self, info: &'a InvocationInfo) -> HookFuture<'a> {
        Box::pin(async move {
            tracing::info!(
                "invocation start, arn: {}, first invocation: {}",
                info.execution_arn(),
                info.is_first_invocation()
            );
            Ok(())
        })
    }

    fn on_invocation_end<'a>(&'a self, info: &'a InvocationEndInfo) -> HookFuture<'a> {
        Box::pin(async move {
            tracing::info!(
                "invocation end, arn: {}, status: {}",
                info.execution_arn(),
                info.status().wire_str()
            );
            Ok(())
        })
    }
}

The SDK awaits every plugin's on_invocation_start before the handler runs. It awaits on_invocation_end once per invocation, after every checkpoint of the invocation is written. info.status() is Succeeded, Failed, Pending, or Retrying, and wire_str() gives SUCCEEDED, FAILED, PENDING, or RETRYING.

Operation hooks¶

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

TypeScriptPythonJavaRust

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

impl Plugin for ExamplePlugin {
    fn on_operation_start(&self, info: &OperationInfo) -> HookWork {
        tracing::info!(
            "operation {} start, type: {:?}, replay: {}",
            info.name().unwrap_or_default(),
            info.kind().map(OperationKind::wire_str),
            info.is_replay()
        );
        Ok(None)
    }

    fn on_operation_end(&self, info: &OperationInfo) -> HookWork {
        tracing::info!(
            "operation {} end, status: {:?}, error: {:?}",
            info.name().unwrap_or_default(),
            info.status().map(OperationStatus::wire_str),
            info.error().map(ToString::to_string)
        );
        Ok(None)
    }
}

info.is_replay() indicates whether the hook is firing for the first time or during a replay. A step replayed from a finished record fires no operation hook. A finished wait, invoke, or create_callback fires on_operation_end on every replay. Its is_replay() is true unless the service changed the operation since the previous invocation. An operation whose .spawn() handle was dropped still fires its end hook if it settles before its scope closes or the invocation ends. Returning from the handler ends the invocation. An operation the SDK stops fires no end hook. These hooks run synchronously and must not block. Put async work in the future they return.

User-function hooks¶

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

TypeScriptPythonJavaRust

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

impl Plugin for ExamplePlugin {
    fn on_user_function_start(&self, info: &OperationInfo) -> HookWork {
        tracing::info!(
            "user function {} start, attempt: {:?}",
            info.name().unwrap_or_default(),
            info.attempt()
        );
        Ok(None)
    }

    fn on_user_function_end(
        &self,
        info: &OperationInfo,
        outcome: Result<(), &WireError>,
    ) -> HookWork {
        tracing::info!(
            "user function {} end, attempt: {:?}, outcome: {:?}",
            info.name().unwrap_or_default(),
            info.attempt(),
            outcome.map_err(ToString::to_string)
        );
        Ok(None)
    }
}

These hooks fire around step attempts, wait-for-condition checks, child context functions, map items and their container, parallel branches and their container, the body of a wait_for_callback, and the join of each combinator. outcome is Ok(()) or the WireError that the SDK records, never the value. Start hooks run in registration order, and end hooks in reverse order. A user function that panics or that the SDK stops before it returns gets no end hook. The hooks run on the task that polls the user function.

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.

TypeScriptPythonJavaRust

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

def on_operation_change(self, info: OperationChangeInfo) -> None:
    print("operations changed", "ids:", list(info.updated_operations.keys()))

@Override
public void onOperationChange(OperationChangeInfo info) {
    System.out.println("operations changed, ids: " + info.updatedOperations().keySet());
}

impl Plugin for ExamplePlugin {
    fn on_operation_change(&self, info: &OperationChangeInfo) -> HookWork {
        let ids: Vec<&str> = info
            .updated_operations()
            .iter()
            .map(OperationInfo::id)
            .collect();
        tracing::info!("operations changed, ids: {ids:?}");
        Ok(None)
    }
}

info.updated_operations() is a slice of OperationInfo, and info.operations() maps every operation ID, EXECUTION included, to its OperationInfo.

Use a plugin¶

Provide plugins in the durable handler configuration. The SDK calls each plugin in the order you provide. The OpenTelemetry plugins and the Durable Execution Telemetry Exporter are ready-made plugins that you register the same way.

TypeScriptPythonJavaRust

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

Register each factory with durable::OptionsBuilder::plugin, and pass the options to durable::run_with_options. This program defines a small plugin that logs each operation start. Register the factory from Define a plugin the same way. build() returns a durable::OptionsValidationError when two factories return the same non-empty exclusive_group().

use aws_durable_execution_sdk as durable;
use durable::plugin::{HookWork, InvocationInfo, OperationInfo, Plugin, PluginFactory};

struct ExamplePlugin;

impl Plugin for ExamplePlugin {
    fn on_operation_start(&self, info: &OperationInfo) -> HookWork {
        tracing::info!("operation {} start", info.name().unwrap_or_default());
        Ok(None)
    }
}

struct ExamplePluginFactory;

impl PluginFactory for ExamplePluginFactory {
    fn create_plugin(
        &self,
        _info: &InvocationInfo,
    ) -> Result<Option<Box<dyn Plugin>>, durable::BoxError> {
        Ok(Some(Box::new(ExamplePlugin)))
    }
}

async fn handler(
    _event: serde_json::Value,
    ctx: durable::DurableContext,
) -> Result<String, durable::BoxError> {
    let result = ctx
        .step(|_| async { Ok("done".to_owned()) })
        .name("process")
        .await?;
    Ok(result)
}

#[tokio::main]
async fn main() -> Result<(), lambda_runtime::Error> {
    durable::init_default_subscriber()?;
    let options = durable::Options::builder()
        .plugin(ExamplePluginFactory)
        .build()?;
    durable::run_with_options(handler, options).await
}

Logging from a plugin¶

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

TypeScriptPythonJavaRust

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

A plugin enters a tracing span that carries its fields around next() in wrap_invocation. The records that the handler, its operations, and the operation hooks write during the invocation descend from that span. The default subscriber writes the span's fields as top-level keys of each record, next to the SDK's fields. To add fields to the records of one user function only, enter the span in wrap_user_function instead. See Logging for more about tracing records.

impl Plugin for ExamplePlugin {
    fn wrap_invocation(&self, _info: &InvocationInfo, next: &mut dyn FnMut()) {
        tracing::info_span!("example_plugin", service = "orders").in_scope(next);
    }
}

Wrap hooks¶

SDK support

The Python and Java SDKs do not have these hooks.

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.

TypeScriptRust

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

wrap_invocation runs around each poll of the invocation, so keep it cheap and do not block in it. A future runs across many polls, on any thread of the runtime, so a plugin enters its context again on each poll. Call next() once to run the rest of the invocation. A wrap that calls next() inside its context, for example with a tracing span's in_scope, makes that context current in the handler and in every operation of the invocation.

impl Plugin for ExamplePlugin {
    fn wrap_invocation(&self, info: &InvocationInfo, next: &mut dyn FnMut()) {
        tracing::info!("wrap invocation, arn: {}", info.execution_arn());
        next();
    }
}

Wrap child context¶

This hook surrounds a single child context function.

TypeScriptRust

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

wrap_user_function runs around each poll of every user function, so keep it cheap and do not block in it. A child context function has the kind durable::OperationKind::Context. Map items and their container, parallel branches and their container, the body of a wait_for_callback, and the join of each combinator have that kind too. info.sub_type() tells them apart, such as MapIteration for a map item.

impl Plugin for ExamplePlugin {
    fn wrap_user_function(&self, info: &OperationInfo, next: &mut dyn FnMut()) {
        if info.kind() == Some(OperationKind::Context) {
            tracing::info!("wrap child context {}", info.name().unwrap_or_default());
        }
        next();
    }
}

Wrap operation attempt¶

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

TypeScriptRust

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

wrap_user_function runs around each poll of every user function, so keep it cheap and do not block in it. A step attempt or a wait-for-condition check has the kind durable::OperationKind::Step and an attempt number.

impl Plugin for ExamplePlugin {
    fn wrap_user_function(&self, info: &OperationInfo, next: &mut dyn FnMut()) {
        if let (Some(OperationKind::Step), Some(attempt)) = (info.kind(), info.attempt()) {
            tracing::info!(
                "wrap attempt {}, attempt: {attempt}",
                info.name().unwrap_or_default()
            );
        }
        next();
    }
}