Struct DurableContext 

pub struct DurableContext { /* private fields */ }

The main context for durable execution operations.

DurableContext provides all the durable operations that user code can use to build reliable workflows. It handles checkpointing, replay, and state management automatically.

Thread Safety

DurableContext is Send + Sync and can be safely shared across async tasks. The internal state uses appropriate synchronization primitives for concurrent access.

Example

async fn my_workflow(ctx: DurableContext) -> Result<String, DurableError> {
    // Execute a step (automatically checkpointed)
    let result = ctx.step(|_| Ok("hello".to_string()), None).await?;
     
    // Wait for 5 seconds (suspends Lambda, resumes later)
    ctx.wait(Duration::from_seconds(5), None).await?;
     
    Ok(result)
}

Implementations

impl DurableContext

pub fn new(state: Arc<ExecutionState>) -> DurableContext

Creates a new DurableContext with the given state.

Arguments

• state - The execution state manager

pub fn from_lambda_context( state: Arc<ExecutionState>, lambda_context: Context, ) -> DurableContext

Creates a DurableContext from a Lambda context.

This is the primary factory method used when running in AWS Lambda.

Arguments

• state - The execution state manager
• lambda_context - The Lambda runtime context

pub fn create_child_context( &self, parent_operation_id: impl Into<String>, ) -> DurableContext

Creates a child context for nested operations.

Child contexts have their own step counter but share the same execution state. Operations in a child context are tracked with the parent’s operation ID.

Arguments

• parent_operation_id - The operation ID of the parent operation

pub fn set_logger(&mut self, logger: Arc<dyn Logger>)

Sets a custom logger for this context.

Arguments

• logger - The logger implementation to use

pub fn with_logger(self, logger: Arc<dyn Logger>) -> DurableContext

Returns a new context with the specified logger.

Arguments

• logger - The logger implementation to use

pub fn configure_logger(&self, logger: Arc<dyn Logger>)

Reconfigures the logger for this context at runtime.

All subsequent log calls on this context (and any clones sharing the same underlying RwLock) will use the new logger.

Arguments

• logger - The new logger implementation to use

pub fn state(&self) -> &Arc<ExecutionState>

Returns a reference to the execution state.

pub fn durable_execution_arn(&self) -> &str

Returns the durable execution ARN.

pub fn parent_id(&self) -> Option<&str>

Returns the parent operation ID, if any.

pub fn lambda_context(&self) -> Option<&Context>

Returns a reference to the Lambda context, if available.

pub fn logger(&self) -> Arc<dyn Logger>

Returns a reference to the logger.

pub fn next_operation_id(&self) -> String

Generates the next operation ID for this context.

This method is thread-safe and will generate unique, deterministic IDs based on the context’s base ID and step counter.

pub fn next_operation_identifier( &self, name: Option<String>, ) -> OperationIdentifier

Creates an OperationIdentifier for the next operation.

Arguments

• name - Optional human-readable name for the operation

pub fn current_step_counter(&self) -> u64

Returns the current step counter value without incrementing.

pub fn create_log_info(&self) -> LogInfo

Creates log info for the current context.

The returned LogInfo includes the current replay status from the execution state, allowing loggers to distinguish between fresh executions and replayed operations.

pub fn create_log_info_with_operation(&self, operation_id: &str) -> LogInfo

Creates log info with an operation ID.

The returned LogInfo includes the current replay status from the execution state, allowing loggers to distinguish between fresh executions and replayed operations.

pub fn create_log_info_with_replay( &self, operation_id: &str, is_replay: bool, ) -> LogInfo

Creates log info with explicit replay status.

This method allows callers to explicitly set the replay status, which is useful when the operation-specific replay status differs from the global execution state replay status.

Arguments

• operation_id - The operation ID to include in the log info
• is_replay - Whether this specific operation is being replayed

pub fn log_info(&self, message: &str)

Logs a message at INFO level with automatic context.

This method automatically includes the durable_execution_arn and parent_id in the log output without requiring the caller to specify them.

Arguments

• message - The message to log

Example

ctx.log_info("Processing order started");

pub fn log_info_with(&self, message: &str, fields: &[(&str, &str)])

Logs a message at INFO level with extra fields.

This method automatically includes the durable_execution_arn and parent_id in the log output, plus any additional fields specified.

Arguments

• message - The message to log
• fields - Additional key-value pairs to include in the log

Example

ctx.log_info_with("Processing order", &[("order_id", "123"), ("amount", "99.99")]);

pub fn log_debug(&self, message: &str)

Logs a message at DEBUG level with automatic context.

This method automatically includes the durable_execution_arn and parent_id in the log output without requiring the caller to specify them.

Arguments

• message - The message to log

Example

ctx.log_debug("Entering validation step");

pub fn log_debug_with(&self, message: &str, fields: &[(&str, &str)])

Logs a message at DEBUG level with extra fields.

This method automatically includes the durable_execution_arn and parent_id in the log output, plus any additional fields specified.

Arguments

• message - The message to log
• fields - Additional key-value pairs to include in the log

Example

ctx.log_debug_with("Variable state", &[("x", "42"), ("y", "100")]);

pub fn log_warn(&self, message: &str)

Logs a message at WARN level with automatic context.

This method automatically includes the durable_execution_arn and parent_id in the log output without requiring the caller to specify them.

Arguments

• message - The message to log

Example

ctx.log_warn("Retry attempt 3 of 5");

pub fn log_warn_with(&self, message: &str, fields: &[(&str, &str)])

Logs a message at WARN level with extra fields.

This method automatically includes the durable_execution_arn and parent_id in the log output, plus any additional fields specified.

Arguments

• message - The message to log
• fields - Additional key-value pairs to include in the log

Example

ctx.log_warn_with("Rate limit approaching", &[("current", "95"), ("limit", "100")]);

pub fn log_error(&self, message: &str)

Logs a message at ERROR level with automatic context.

This method automatically includes the durable_execution_arn and parent_id in the log output without requiring the caller to specify them.

Arguments

• message - The message to log

Example

ctx.log_error("Failed to process payment");

pub fn log_error_with(&self, message: &str, fields: &[(&str, &str)])

Logs a message at ERROR level with extra fields.

This method automatically includes the durable_execution_arn and parent_id in the log output, plus any additional fields specified.

Arguments

• message - The message to log
• fields - Additional key-value pairs to include in the log

Example

ctx.log_error_with("Payment failed", &[("error_code", "INSUFFICIENT_FUNDS"), ("amount", "150.00")]);

pub fn get_original_input<T>(&self) -> Result<T, DurableError>

where T: DeserializeOwned,

Returns the original user input from the EXECUTION operation.

This method deserializes the input payload from the EXECUTION operation’s ExecutionDetails.InputPayload field into the requested type.

Type Parameters

• T - The type to deserialize the input into. Must implement DeserializeOwned.

Returns

Ok(T) if the input exists and can be deserialized, or a DurableError if:

• No EXECUTION operation exists
• No input payload is available
• Deserialization fails

Example

#[derive(Deserialize)]
struct OrderEvent {
    order_id: String,
    amount: f64,
}

async fn my_workflow(ctx: DurableContext) -> Result<(), DurableError> {
    // Get the original input that started this execution
    let event: OrderEvent = ctx.get_original_input()?;
    println!("Processing order: {}", event.order_id);
    Ok(())
}

pub fn get_original_input_raw(&self) -> Option<&str>

Returns the raw original user input as a string, if available.

This method returns the raw JSON string from the EXECUTION operation’s ExecutionDetails.InputPayload field without deserializing it.

Returns

Some(&str) if the input exists, None otherwise.

pub async fn complete_execution_success<T>( &self, result: &T, ) -> Result<(), DurableError>

where T: Serialize,

Completes the execution with a successful result via checkpointing.

This method checkpoints a SUCCEED action on the EXECUTION operation, which is useful for large results that exceed the Lambda response size limit (6MB). After calling this method, the Lambda function should return an empty result.

Arguments

• result - The result to checkpoint. Must implement Serialize.

Returns

Ok(()) on success, or a DurableError if:

• No EXECUTION operation exists
• Serialization fails
• The checkpoint fails

Example

async fn my_workflow(ctx: DurableContext) -> Result<(), DurableError> {
    let large_result = compute_large_result().await?;
     
    // Check if result would exceed Lambda response limit
    if DurableExecutionInvocationOutput::would_exceed_max_size(&large_result) {
        // Checkpoint the result instead of returning it
        ctx.complete_execution_success(&large_result).await?;
        // Return empty result - the actual result is checkpointed
        return Ok(());
    }
     
    Ok(())
}

pub async fn complete_execution_failure( &self, error: ErrorObject, ) -> Result<(), DurableError>

Completes the execution with a failure via checkpointing.

This method checkpoints a FAIL action on the EXECUTION operation. After calling this method, the Lambda function should return a FAILED status.

Arguments

• error - The error details to checkpoint

Returns

Ok(()) on success, or a DurableError if:

• No EXECUTION operation exists
• The checkpoint fails

Example

async fn my_workflow(ctx: DurableContext) -> Result<(), DurableError> {
    if let Err(e) = process_order().await {
        // Checkpoint the failure
        ctx.complete_execution_failure(ErrorObject::new("ProcessingError", &e.to_string())).await?;
        return Err(DurableError::execution(&e.to_string()));
    }
    Ok(())
}

pub async fn complete_execution_if_large<T>( &self, result: &T, ) -> Result<bool, DurableError>

where T: Serialize,

Completes the execution with a successful result, automatically handling large results.

This method checks if the result would exceed the Lambda response size limit (6MB). If so, it checkpoints the result via the EXECUTION operation and returns true. If the result fits within the limit, it returns false and the caller should return the result normally.

Arguments

• result - The result to potentially checkpoint. Must implement Serialize.

Returns

Ok(true) if the result was checkpointed (caller should return empty result), Ok(false) if the result fits within limits (caller should return it normally), or a DurableError if checkpointing fails.

Example

async fn my_workflow(ctx: DurableContext) -> Result<LargeResult, DurableError> {
    let result = compute_result().await?;
     
    // Automatically handle large results
    if ctx.complete_execution_if_large(&result).await? {
        // Result was checkpointed, return a placeholder
        // The actual result is stored in the EXECUTION operation
        return Ok(LargeResult::default());
    }
     
    // Result fits within limits, return normally
    Ok(result)
}

pub async fn step<T, F>( &self, func: F, config: Option<StepConfig>, ) -> Result<T, DurableError>

where T: DurableValue, F: StepFn<T>,

Executes a step operation with automatic checkpointing.

Steps are the fundamental unit of work in durable executions. Each step is checkpointed, allowing the workflow to resume from the last completed step after interruptions.

Arguments

• func - The function to execute
• config - Optional step configuration (retry strategy, semantics, serdes)

Returns

The result of the step function, or an error if execution fails.

Example

let result: i32 = ctx.step(|_step_ctx| {
    Ok(42)
}, None).await?;

pub async fn step_named<T, F>( &self, name: &str, func: F, config: Option<StepConfig>, ) -> Result<T, DurableError>

where T: DurableValue, F: StepFn<T>,

Executes a named step operation with automatic checkpointing.

Same as step, but allows specifying a human-readable name for the operation.

Arguments

• name - Human-readable name for the step
• func - The function to execute
• config - Optional step configuration

Example

let result: i32 = ctx.step_named("validate_input", |_step_ctx| {
    Ok(42)
}, None).await?;

pub async fn wait( &self, duration: Duration, name: Option<&str>, ) -> Result<(), DurableError>

Pauses execution for a specified duration.

Wait operations suspend the Lambda execution and resume after the specified duration has elapsed. This is efficient because it doesn’t block Lambda resources during the wait.

Arguments

• duration - The duration to wait (must be at least 1 second)
• name - Optional human-readable name for the operation

Returns

Ok(()) when the wait has elapsed, or an error if validation fails.

Example

// Wait for 5 seconds
ctx.wait(Duration::from_seconds(5), None).await?;

// Wait with a name
ctx.wait(Duration::from_minutes(1), Some("wait_for_processing")).await?;

pub async fn cancel_wait(&self, operation_id: &str) -> Result<(), DurableError>

Cancels an active wait operation.

This method allows cancelling a wait operation that was previously started. If the wait has already completed (succeeded, failed, or timed out), this method will return Ok(()) without making any changes.

Arguments

• operation_id - The operation ID of the wait to cancel

Returns

Ok(()) if the wait was cancelled or was already completed, or an error if:

• The operation doesn’t exist
• The operation is not a WAIT operation
• The checkpoint fails

Example

// Start a wait in a parallel branch
let wait_op_id = ctx.next_operation_id();

// Later, cancel the wait from another branch
ctx.cancel_wait(&wait_op_id).await?;

pub async fn create_callback<T>( &self, config: Option<CallbackConfig>, ) -> Result<Callback<T>, DurableError>

where T: Serialize + DeserializeOwned,

Creates a callback and returns a handle to wait for the result.

Callbacks allow external systems to signal completion of asynchronous operations. The callback ID can be shared with external systems, which can then call the Lambda durable execution callback API.

Arguments

• config - Optional callback configuration (timeout, heartbeat)

Returns

A Callback<T> handle that can be used to wait for the result.

Example

let callback = ctx.create_callback::<ApprovalResponse>(None).await?;

// Share callback.callback_id with external system
notify_approver(&callback.callback_id).await?;

// Wait for the callback result
let approval = callback.result().await?;

pub async fn create_callback_named<T>( &self, name: &str, config: Option<CallbackConfig>, ) -> Result<Callback<T>, DurableError>

where T: Serialize + DeserializeOwned,

Creates a named callback and returns a handle to wait for the result.

Same as create_callback, but allows specifying a human-readable name.

pub async fn invoke<P, R>( &self, function_name: &str, payload: P, config: Option<InvokeConfig<P, R>>, ) -> Result<R, DurableError>

where P: Serialize + DeserializeOwned + Send, R: Serialize + DeserializeOwned + Send,

Invokes another durable Lambda function.

This method calls another Lambda function and waits for its result. The invocation is checkpointed, so if the workflow is interrupted, it will resume with the result of the invocation.

Arguments

• function_name - The name or ARN of the Lambda function to invoke
• payload - The payload to send to the function
• config - Optional invoke configuration (timeout, serdes)

Returns

The result from the invoked function, or an error if invocation fails.

Example

let result: ProcessingResult = ctx.invoke(
    "process-order-function",
    OrderPayload { order_id: "123".to_string() },
    None,
).await?;

pub async fn map<T, U, F, Fut>( &self, items: Vec<T>, func: F, config: Option<MapConfig>, ) -> Result<BatchResult<U>, DurableError>

where T: Serialize + DeserializeOwned + Send + Sync + Clone + 'static, U: Serialize + DeserializeOwned + Send + 'static, F: Fn(DurableContext, T, usize) -> Fut + Send + Sync + Clone + 'static, Fut: Future<Output = Result<U, DurableError>> + Send + 'static,

Processes a collection in parallel with configurable concurrency.

Map operations execute a function for each item in the collection, with configurable concurrency limits and failure tolerance.

Arguments

• items - The collection of items to process
• func - The function to apply to each item
• config - Optional map configuration (concurrency, completion criteria)

Returns

A BatchResult<U> containing results for all items.

Example

let results = ctx.map(
    vec![1, 2, 3, 4, 5],
    |child_ctx, item, index| async move {
        Ok(item * 2)
    },
    Some(MapConfig {
        max_concurrency: Some(3),
        ..Default::default()
    }),
).await?;

pub async fn parallel<T, F, Fut>( &self, branches: Vec<F>, config: Option<ParallelConfig>, ) -> Result<BatchResult<T>, DurableError>

where T: Serialize + DeserializeOwned + Send + 'static, F: FnOnce(DurableContext) -> Fut + Send + 'static, Fut: Future<Output = Result<T, DurableError>> + Send + 'static,

Executes multiple operations in parallel.

Parallel operations execute multiple independent functions concurrently, with configurable concurrency limits and completion criteria.

Arguments

• branches - The list of functions to execute in parallel
• config - Optional parallel configuration (concurrency, completion criteria)

Returns

A BatchResult<T> containing results for all branches.

Example

let results = ctx.parallel(
    vec![
        |ctx| async move { Ok(fetch_data_a(&ctx).await?) },
        |ctx| async move { Ok(fetch_data_b(&ctx).await?) },
        |ctx| async move { Ok(fetch_data_c(&ctx).await?) },
    ],
    None,
).await?;

pub async fn run_in_child_context<T, F, Fut>( &self, func: F, config: Option<ChildConfig>, ) -> Result<T, DurableError>

where T: Serialize + DeserializeOwned + Send, F: FnOnce(DurableContext) -> Fut + Send, Fut: Future<Output = Result<T, DurableError>> + Send,

Executes a function in a child context.

Child contexts provide isolation for nested workflows. Operations in a child context are tracked separately and can be checkpointed as a unit.

Arguments

• func - The function to execute in the child context
• config - Optional child context configuration

Returns

The result of the child function, or an error if execution fails.

Example

let result = ctx.run_in_child_context(|child_ctx| async move {
    let step1 = child_ctx.step(|_| Ok(1), None).await?;
    let step2 = child_ctx.step(|_| Ok(2), None).await?;
    Ok(step1 + step2)
}, None).await?;

pub async fn run_in_child_context_named<T, F, Fut>( &self, name: &str, func: F, config: Option<ChildConfig>, ) -> Result<T, DurableError>

where T: Serialize + DeserializeOwned + Send, F: FnOnce(DurableContext) -> Fut + Send, Fut: Future<Output = Result<T, DurableError>> + Send,

Executes a named function in a child context.

Same as run_in_child_context, but allows specifying a human-readable name.

pub async fn wait_for_condition<T, S, F>( &self, check: F, config: WaitForConditionConfig<S>, ) -> Result<T, DurableError>

where T: Serialize + DeserializeOwned + Send, S: Serialize + DeserializeOwned + Clone + Send + Sync, F: Fn(&S, &WaitForConditionContext) -> Result<T, Box<dyn Error + Send + Sync>> + Send + Sync,

Polls until a condition is met.

This method repeatedly checks a condition until it returns a successful result. Between checks, it waits for a configurable duration using the RETRY mechanism with NextAttemptDelaySeconds.

Implementation

This method is implemented as a single STEP operation with RETRY mechanism, which is more efficient than using multiple steps and waits. The state is passed as Payload on retry (not Error), and the attempt number is tracked in StepDetails.Attempt.

Arguments

• check - The function to check the condition
• config - Configuration for the wait (interval, max attempts, timeout)

Returns

The result when the condition is met, or an error if timeout/max attempts exceeded.

Example

let result = ctx.wait_for_condition(
    |state, ctx| {
        // Check if order is ready
        let status = check_order_status(&state.order_id)?;
        if status == "ready" {
            Ok(OrderReady { order_id: state.order_id.clone() })
        } else {
            Err("Order not ready yet".into())
        }
    },
    WaitForConditionConfig::from_interval(
        OrderState { order_id: "123".to_string() },
        Duration::from_seconds(5),
        Some(10),
    ),
).await?;

pub async fn wait_for_callback<T, F, Fut>( &self, submitter: F, config: Option<CallbackConfig>, ) -> Result<T, DurableError>

where T: Serialize + DeserializeOwned + Send + Sync, F: FnOnce(String) -> Fut + Send + 'static, Fut: Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'static,

Creates a callback and waits for the result with a submitter function.

This is a convenience method that combines callback creation with a submitter function that sends the callback ID to an external system. The submitter execution is checkpointed within a child context to ensure replay safety - the submitter will not be re-executed during replay.

Arguments

• submitter - Function that receives the callback ID and submits it to external system
• config - Optional callback configuration (timeout, heartbeat)

Returns

The callback result from the external system.

Example

let approval = ctx.wait_for_callback(
    |callback_id| async move {
        // Send callback ID to approval system
        send_approval_request(&callback_id, &request).await
    },
    Some(CallbackConfig {
        timeout: Duration::from_hours(24),
        ..Default::default()
    }),
).await?;

pub async fn all<T, Fut>( &self, futures: Vec<Fut>, ) -> Result<Vec<T>, DurableError>

where T: Serialize + DeserializeOwned + Send + Clone + 'static, Fut: Future<Output = Result<T, DurableError>> + Send + 'static,

Waits for all futures to complete successfully.

Returns all results if all futures succeed, or returns the first error encountered. This is implemented within a STEP operation for durability.

Arguments

• futures - Vector of futures to execute

Returns

Ok(Vec<T>) if all futures succeed, or Err with the first error.

Example

let results = ctx.all(vec![
    ctx.step(|_| Ok(1), None),
    ctx.step(|_| Ok(2), None),
    ctx.step(|_| Ok(3), None),
]).await?;
assert_eq!(results, vec![1, 2, 3]);

pub async fn all_settled<T, Fut>( &self, futures: Vec<Fut>, ) -> Result<BatchResult<T>, DurableError>

where T: Serialize + DeserializeOwned + Send + Clone + 'static, Fut: Future<Output = Result<T, DurableError>> + Send + 'static,

Waits for all futures to settle (success or failure).

Returns a BatchResult containing outcomes for all futures, regardless of success or failure. This is implemented within a STEP operation for durability.

Arguments

• futures - Vector of futures to execute

Returns

BatchResult<T> containing results for all futures.

Example

let results = ctx.all_settled(vec![
    ctx.step(|_| Ok(1), None),
    ctx.step(|_| Err("error".into()), None),
    ctx.step(|_| Ok(3), None),
]).await?;
assert_eq!(results.success_count(), 2);
assert_eq!(results.failure_count(), 1);

pub async fn race<T, Fut>(&self, futures: Vec<Fut>) -> Result<T, DurableError>

where T: Serialize + DeserializeOwned + Send + Clone + 'static, Fut: Future<Output = Result<T, DurableError>> + Send + 'static,

Returns the result of the first future to settle.

Returns the result (success or failure) of whichever future completes first. This is implemented within a STEP operation for durability.

Arguments

• futures - Vector of futures to execute

Returns

The result of the first future to settle.

Example

let result = ctx.race(vec![
    ctx.step(|_| Ok(1), None),
    ctx.step(|_| Ok(2), None),
]).await?;
// result is either 1 or 2, whichever completed first

pub async fn any<T, Fut>(&self, futures: Vec<Fut>) -> Result<T, DurableError>

where T: Serialize + DeserializeOwned + Send + Clone + 'static, Fut: Future<Output = Result<T, DurableError>> + Send + 'static,

Returns the result of the first future to succeed.

Returns the result of the first future to succeed. If all futures fail, returns an error containing all the failures. This is implemented within a STEP operation for durability.

Arguments

• futures - Vector of futures to execute

Returns

The result of the first future to succeed, or an error if all fail.

Example

let result = ctx.any(vec![
    ctx.step(|_| Err("error".into()), None),
    ctx.step(|_| Ok(2), None),
    ctx.step(|_| Ok(3), None),
]).await?;
// result is either 2 or 3, whichever succeeded first

Trait Implementations

impl Clone for DurableContext

fn clone(&self) -> DurableContext

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for DurableContext

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.