Wait for Condition¶

The check receives the durable.StepContext first, then the state, and returns the next state. The stop decision is in the strategy, not the check.

package main

import (
    "time"

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

type JobState struct {
    JobID  string `json:"jobId"`
    Status string `json:"status"`
    Done   bool   `json:"done"`
}

// getJobStatus stands in for a call to the job service. It reports the job
// complete on the second check.
func getJobStatus(sc durable.StepContext, jobID string) (string, error) {
    if sc.Attempt() < 2 {
        return "RUNNING", nil
    }
    return "COMPLETED", nil
}

func handler(ctx durable.Context, event map[string]any) (JobState, error) {
    // Poll until the job completes.
    return durable.WaitForCondition(ctx, "wait_for_job",
        func(sc durable.StepContext, state JobState) (JobState, error) {
            status, err := getJobStatus(sc, state.JobID)
            if err != nil {
                return state, err
            }
            state.Status = status
            state.Done = status == "COMPLETED"
            return state, nil
        },
        durable.ConditionConfig[JobState]{
            InitialState: JobState{JobID: "job-123", Status: "pending", Done: false},
            WaitStrategy: func(state JobState, attempt int) durable.WaitDecision {
                if state.Done {
                    return durable.WaitDecision{Continue: false}
                }
                return durable.WaitDecision{
                    Continue: true,
                    Delay:    time.Duration(min(5*(1<<attempt), 300)) * time.Second,
                }
            },
        },
    )
}

func main() {
    durable.Start(handler)
}

The check receives the durable::StepContext first, then the state, and returns the next state. The wait strategy decides when to stop.

use std::time::Duration;

use aws_durable_execution_sdk as durable;
use durable::builders::wait_for_condition::WaitDecision;
use serde::{Deserialize, Serialize};

#[derive(Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct JobState {
    job_id: String,
    status: String,
    done: bool,
}

fn get_job_status(step_ctx: &durable::StepContext, _job_id: &str) -> String {
    let status = if step_ctx.attempt() < 2 {
        "RUNNING"
    } else {
        "COMPLETED"
    };
    status.to_owned()
}

async fn handler(
    _event: serde_json::Value,
    ctx: durable::DurableContext,
) -> Result<JobState, durable::BoxError> {
    let result = ctx
        .wait_for_condition(
            |step_ctx, state: JobState| async move {
                let status = get_job_status(&step_ctx, &state.job_id);
                let done = status == "COMPLETED";
                Ok(JobState {
                    status,
                    done,
                    ..state
                })
            },
            JobState {
                job_id: "job-123".to_owned(),
                status: "pending".to_owned(),
                done: false,
            },
        )
        .name("wait_for_job")
        .wait_strategy_fn(|state: JobState, attempt| {
            if state.done {
                return WaitDecision::complete();
            }
            WaitDecision::continue_with(Duration::from_secs(
                5_u64.saturating_mul(2_u64.saturating_pow(attempt)).min(300),
            ))
        })
        .await?;
    Ok(result)
}

#[tokio::main]
async fn main() -> Result<(), lambda_runtime::Error> {
    durable::init_default_subscriber()?;
    durable::run(handler).await
}

The configuration is a ConditionConfig[S] struct passed positionally. cfg.WaitStrategy is required. A nil strategy makes WaitForCondition return an error at the call, before any check runs.

In a check, StepContext.Attempt() returns the 1-based poll attempt, the same number the wait strategy receives. A check that returns an error fails the operation at once, with no retry. WaitForCondition returns a *durable.WaitForConditionError when the check returns an error and when the strategy stops with Err set. A negative Delay makes it return an error.

func WaitForCondition[S any](ctx Context, name string, check func(StepContext, S) (S, error), cfg ConditionConfig[S]) (S, error)

type ConditionConfig[S any] struct {
    // InitialState is the state passed to the first check.
    InitialState S

    // WaitStrategy decides, after each check, whether to keep waiting and
    // for how long. attempt is the 1-based number of completed checks.
    // The strategy must be a deterministic function of its arguments,
    // except for randomized jitter in the returned delay.
    //
    // WaitStrategy is required and must not be nil. When it is nil,
    // [WaitForCondition] returns an error at the call and runs no check.
    WaitStrategy func(state S, attempt int) WaitDecision

    // Serdes overrides the serializer for the condition state.
    Serdes Serdes
}

initial_state is the second argument of wait_for_condition. Set the operation name with .name(...). The name is part of replay identity, so it must stay the same each time the handler replays. Set the strategy with .wait_strategy(...), which takes a WaitStrategy, or with .wait_strategy_fn(...), which takes a function. A strategy is required. Only the builder that one of the two methods returns can be awaited or turned into a future with .future() or .spawn(), so code without a strategy fails to compile. A second call replaces the first strategy. The SDK round-trips the initial state through the serdes before the first check. So with the default durable::JsonSerdes, the state type needs Clone, Serialize, and DeserializeOwned.

In a check, StepContext::attempt() returns the 1-based poll attempt, the same number the wait strategy receives. A check that returns Err fails the operation at once, without a call to the strategy. .await returns a durable::OperationError of kind OperationErrorKind::WaitForCondition, whose kind() is a durable::WaitForConditionErrorKind:

• CheckFailed when the check returns Err.
• MaxChecksExceeded(details) when the strategy returns WaitDecision::exhausted(reason). details.checks() returns the number of checks performed.
• SerializationFailed when the state cannot be serialized or deserialized.

impl DurableContext {
    // ...
    pub fn wait_for_condition<S, F, Fut>(
        &self,
        check: F,
        initial_state: S,
    ) -> WaitForConditionBuilder<S, F, Fut>
    where
        F: Fn(StepContext, S) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<S, BoxError>> + Send + 'static,
        S: Clone + Send + Sync + 'static;
    // ...
}
// ...

pub struct WaitForConditionBuilder<S, F, Fut, SD = JsonSerdes, WS = NoWaitStrategy> {
    // ...
}

impl<S, F, Fut, SD, WS> WaitForConditionBuilder<S, F, Fut, SD, WS>
where
    S: Clone + Send + Sync + 'static,
    F: Fn(StepContext, S) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = Result<S, BoxError>> + Send + 'static,
{
    pub fn name(mut self, name: impl Into<String>) -> Self;
    // ...
    pub fn wait_strategy(
        self,
        strategy: WaitStrategy<S>,
    ) -> WaitForConditionBuilder<S, F, Fut, SD, WithWaitStrategy<S>>;
    // ...
    pub fn wait_strategy_fn<W>(
        self,
        strategy: W,
    ) -> WaitForConditionBuilder<S, F, Fut, SD, WithWaitStrategy<S>>
    where
        W: Fn(S, u32) -> WaitDecision + Send + Sync + 'static;
    // ...
    pub fn serdes<SD2>(self, serdes: SD2) -> WaitForConditionBuilder<S, F, Fut, SD2, WS>
    where
        SD2: Serdes<S>;
    // ...
}

impl<S, F, Fut, SD> WaitForConditionBuilder<S, F, Fut, SD, WithWaitStrategy<S>>
where
    S: Clone + Send + Sync + 'static,
    F: Fn(StepContext, S) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = Result<S, BoxError>> + Send + 'static,
    SD: Serdes<S>,
{
    pub fn future(self) -> DurableFuture<S>;
    // ...
    pub fn spawn(self) -> DurableFuture<S>;
    // ...
}

impl<S, F, Fut, SD> IntoFuture for WaitForConditionBuilder<S, F, Fut, SD, WithWaitStrategy<S>>
// ...
{
    type Output = Result<S, OperationError>;
    // ...
}

The strategy returns a WaitDecision struct. To poll again, set Continue and a Delay. To stop and succeed, leave Continue false and Err nil. To stop and fail, leave Continue false and set Err.

type WaitStrategy[S any] func(state S, attempt int) WaitDecision

type WaitDecision struct {
    // Continue indicates whether to keep waiting and check again.
    Continue bool

    // Delay is how long to suspend before the next check. A Delay of 0 is
    // raised to one second. A positive Delay is rounded up to whole
    // seconds, so a positive Delay under one second waits one second. A
    // negative Delay makes WaitForCondition return an error. Delay is
    // ignored when Continue is false.
    Delay time.Duration

    // Err, when non-nil and Continue is false, signals that the
    // operation should fail with this error rather than succeed. Use it
    // to implement max-attempts or timeout strategies.
    Err error
}

The strategy receives the state and the 1-based attempt, and returns a durable::builders::wait_for_condition::WaitDecision.

• WaitDecision::complete() stops polling and returns the state.
• WaitDecision::continue_with(delay) polls again after delay, rounded up to whole seconds with a minimum of one second.
• WaitDecision::exhausted(reason) stops polling and fails the operation with WaitForConditionErrorKind::MaxChecksExceeded. The record carries the error type "WaitForCondition.MaxChecksExceeded" and the message reason.

pub fn wait_strategy_fn<W>(
    self,
    strategy: W,
) -> WaitForConditionBuilder<S, F, Fut, SD, WithWaitStrategy<S>>
where
    W: Fn(S, u32) -> WaitDecision + Send + Sync + 'static;
// ...

pub enum WaitDecision {
    Complete,
    #[non_exhaustive]
    Continue {
        delay: Duration,
    },
    #[non_exhaustive]
    Exhausted {
        reason: String,
    },
}

impl WaitDecision {
    pub fn complete() -> Self;
    // ...
    pub fn continue_with(delay: Duration) -> Self;
    // ...
    pub fn exhausted(reason: impl Into<String>) -> Self;
    // ...
}

In Java, the stop/continue decision lives in the check function (via WaitForConditionResult). The strategy only computes the delay.

WaitForConditionWaitStrategy<Map<String, String>> strategy = (state, attempt) -> {
    if (attempt >= 10) {
        throw new WaitForConditionFailedException("Max attempts exceeded");
    }
    return Duration.ofSeconds(attempt * 5);
};

To stop and fail, return a WaitDecision with Err set and Continue false, or return an error from the check.

strategy := func(state JobState, attempt int) durable.WaitDecision {
    if state.Status == "COMPLETED" {
        return durable.WaitDecision{Continue: false}
    }
    if attempt >= 10 {
        return durable.WaitDecision{Continue: false, Err: errors.New("max attempts exceeded")}
    }
    return durable.WaitDecision{Continue: true, Delay: time.Duration(attempt) * 5 * time.Second}
}

To stop and fail, return WaitDecision::exhausted(reason) from the strategy, or return Err from the check. Pass the strategy with .wait_strategy_fn(strategy()). JobState is the state type from the walkthrough.

use std::time::Duration;

use aws_durable_execution_sdk as durable;
use durable::builders::wait_for_condition::WaitDecision;
use serde::{Deserialize, Serialize};

#[derive(Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct JobState {
    job_id: String,
    status: String,
    done: bool,
}

fn strategy() -> impl Fn(JobState, u32) -> WaitDecision {
    |state, attempt| {
        if state.status == "COMPLETED" {
            return WaitDecision::complete();
        }
        if attempt >= 10 {
            return WaitDecision::exhausted("Max attempts exceeded");
        }
        WaitDecision::continue_with(Duration::from_secs(u64::from(attempt) * 5))
    }
}

Implement IWaitStrategy<TState> or wrap a delegate with WaitStrategy.FromDelegate. The strategy returns a WaitDecision: WaitDecision.Stop() to stop polling or WaitDecision.ContinueAfter(delay) to poll again.

using Amazon.Lambda.DurableExecution;

public record JobState(string Status);

public static class CustomWaitStrategy
{
    // Full control over polling: the strategy decides whether to continue or
    // stop, and how long to wait before the next attempt.
    public static IWaitStrategy<JobState> Create() =>
        WaitStrategy.FromDelegate<JobState>((state, attempt) =>
        {
            if (state.Status == "COMPLETED")
            {
                return WaitDecision.Stop();
            }
            if (attempt >= 10)
            {
                throw new WaitForConditionException("Max attempts exceeded");
            }
            return WaitDecision.ContinueAfter(TimeSpan.FromSeconds(attempt * 5));
        });
}

import java.time.Duration;
import java.util.Map;
import software.amazon.lambda.durable.retry.JitterStrategy;
import software.amazon.lambda.durable.retry.WaitStrategies;

var defaultStrategy = WaitStrategies.<Map<String, String>>defaultStrategy();

var exponentialStrategy = WaitStrategies.<Map<String, String>>exponentialBackoff(
        10,                          // maxAttempts
        Duration.ofSeconds(5),       // initialDelay
        Duration.ofMinutes(5),       // maxDelay
        2.0,                         // backoffRate
        JitterStrategy.FULL);        // jitter

var fixedDelayStrategy = WaitStrategies.<Map<String, String>>fixedDelay(
        10,                          // maxAttempts
        Duration.ofSeconds(30));     // fixedDelay

WaitStrategies.defaultStrategy() returns the same strategy as WaitStrategies.Presets.DEFAULT: 60 maximum attempts, a 5-second initial delay, a 5-minute maximum delay, a 1.5x backoff rate, and full jitter.

Use WaitStrategies.exponentialBackoff(...) to customize each backoff parameter. Use WaitStrategies.fixedDelay(maxAttempts, delay) for a constant delay between checks. Both factories count attempts from 1 and throw WaitForConditionFailedException when the check remains incomplete after maxAttempts.

Java's strategy factories do not accept shouldContinuePolling. The check function controls completion by returning WaitForConditionResult.stopPolling(...) or WaitForConditionResult.continuePolling(...).

strategy := durable.MustNewWaitStrategy(durable.WaitConfig[JobState]{
    MaxAttempts:    10,
    InitialDelay:   5 * time.Second,
    MaxDelay:       5 * time.Minute,
    BackoffRate:    2.0,
    Jitter:         durable.JitterFull,
    ShouldContinue: func(state JobState) bool { return state.Status != "COMPLETED" },
})

ShouldContinue is the shouldContinuePolling predicate. It returns false when the condition is met. When ShouldContinue is nil, the strategy never stops early, so the operation can only fail at MaxAttempts. The jitter constants are durable.JitterFull, durable.JitterHalf, and durable.JitterNone. For a constant delay, set BackoffRate to 1.

A zero WaitConfig field selects the default listed below. So MaxAttempts: 0 means 60 checks. For one check, set MaxAttempts to 1. InitialDelay and MaxDelay must be at least one second when set. BackoffRate must be finite and not negative. Jitter must be one of the jitter constants. durable.NewWaitStrategy returns an error for an invalid configuration, and durable.MustNewWaitStrategy panics. When the strategy reaches MaxAttempts with the condition unmet, the operation fails with a *durable.WaitForConditionError.

use std::time::Duration;

use aws_durable_execution_sdk as durable;
use durable::builders::JitterStrategy;
use durable::builders::wait_for_condition::WaitStrategy;
use serde::{Deserialize, Serialize};

#[derive(Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct JobState {
    job_id: String,
    status: String,
    done: bool,
}

fn strategy() -> WaitStrategy<JobState> {
    WaitStrategy::builder(|state: &JobState| state.status == "COMPLETED")
        .max_attempts(10)
        .initial_delay(Duration::from_secs(5))
        .max_delay(Duration::from_mins(5))
        .backoff_rate(2.0)
        .jitter(JitterStrategy::Full)
        .build()
}

WaitStrategy::builder takes a completion predicate in place of shouldContinuePolling. The predicate returns true when the condition is met, and polling stops when it does. Pass the built strategy to .wait_strategy(strategy()). The jitter values are durable::builders::JitterStrategy::Full, Half, and None.

A method you do not call keeps the default listed below. The SDK does not check the values. A computed delay is rounded to the nearest second, with a minimum of one second. When the strategy reaches max_attempts with the condition unmet, the operation fails with WaitForConditionErrorKind::MaxChecksExceeded.