Lambda Durable Functions for S3 Ingestion: Pipelines That Wait Without Paying for Compute

{
  "source": ["aws.s3"],
  "detail-type": ["Object Created"],
  "detail": {
    "bucket": { "name": ["example-file-drops"] },
    "object": { "key": [{ "wildcard": "incoming/*/manifest.json" }] }
  }
}

# dispatch-batch: a plain Lambda function (not durable), the target of the EventBridge rule.
# It starts one durable execution per manifest event, and only one, however often
# the event is delivered.
import hashlib, json, os
import boto3

lam = boto3.client("lambda")
INGEST_FN = os.environ["INGEST_FN"]   # qualified: arn:aws:lambda:...:function:ingest-batch:prod

def handler(event, context):
    obj = event["detail"]["object"]
    job = {"bucket": event["detail"]["bucket"]["name"], "key": obj["key"],
           "etag": obj["etag"], "sequencer": obj["sequencer"]}

    # The same S3 event always produces the same name and the same payload, so a
    # duplicate delivery returns the existing execution instead of starting another.
    digest = hashlib.sha256(f"{job['bucket']}/{job['key']}/{job['sequencer']}".encode()).hexdigest()
    try:
        lam.invoke(FunctionName=INGEST_FN, InvocationType="Event",
                   DurableExecutionName=f"batch-{digest[:58]}",     # 1-64 chars: [a-zA-Z0-9-_]
                   Payload=json.dumps(job, sort_keys=True))
    except lam.exceptions.DurableExecutionAlreadyStartedException:
        pass   # this name already ran with a different payload; nothing new to start

# ingest-batch: a Lambda durable function (Python 3.13+, durable execution enabled).
# One execution per partner batch: wait for its files, validate them, load them with
# Glue, and record the load. Every context.* call below is checkpointed.
import json, os
import boto3
from botocore.exceptions import ClientError
from aws_durable_execution_sdk_python import DurableContext, StepContext, durable_execution, durable_step
from aws_durable_execution_sdk_python.config import (
    Duration, MapConfig, StepConfig, StepSemantics, WaitForCallbackConfig)
from aws_durable_execution_sdk_python.retries import RetryStrategyConfig, create_retry_strategy
from aws_durable_execution_sdk_python.waits import WaitForConditionConfig, WaitForConditionDecision

s3, glue, sns = boto3.client("s3"), boto3.client("glue"), boto3.client("sns")
GLUE_JOB = os.environ["GLUE_JOB"]
OWNERS = os.environ["OWNERS_TOPIC_ARN"]
HEADER = os.environ["EXPECTED_HEADER"]          # e.g. "order_id,customer_id,amount,currency"
GLUE_DONE = {"SUCCEEDED", "FAILED", "STOPPED", "TIMEOUT", "ERROR", "EXPIRED"}


# ---- steps: ordinary code whose result is checkpointed ------------------------------

@durable_step
def read_manifest(step: StepContext, bucket: str, key: str) -> dict:
    manifest = json.loads(s3.get_object(Bucket=bucket, Key=key)["Body"].read())
    prefix = key.rsplit("/", 1)[0] + "/"
    return {"bucket": bucket, "prefix": prefix, "expected": [prefix + f for f in manifest["files"]]}


@durable_step
def validate(step: StepContext, bucket: str, key: str) -> list[str]:
    size = s3.head_object(Bucket=bucket, Key=key)["ContentLength"]
    first = s3.get_object(Bucket=bucket, Key=key, Range="bytes=0-4095")["Body"].read()
    header = first.split(b"\n", 1)[0].decode("utf-8-sig").strip()
    problems = []
    if size < 100:
        problems.append(f"{key}: only {size} bytes")
    if header != HEADER:
        problems.append(f"{key}: unexpected header {header[:120]!r}")
    return problems


@durable_step
def start_glue(step: StepContext, batch: dict) -> str:
    run = glue.start_job_run(JobName=GLUE_JOB, Arguments={
        "--input_prefix": f"s3://{batch['bucket']}/{batch['prefix']}"})
    return run["JobRunId"]


@durable_step
def commit(step: StepContext, batch: dict, run_id: str) -> None:
    # The marker downstream jobs wait for. IfNoneMatch makes it write-once, so a step
    # that re-runs after a lost checkpoint can't overwrite it.
    try:
        s3.put_object(Bucket=batch["bucket"], Key=batch["prefix"] + "_LOADED.json",
                      Body=json.dumps({"glue_run_id": run_id, "files": batch["expected"]}),
                      IfNoneMatch="*")
    except ClientError as e:
        if e.response["Error"]["Code"] != "PreconditionFailed":
            raise


def notify(subject: str, detail) -> None:
    sns.publish(TopicArn=OWNERS, Subject=subject[:100], Message=json.dumps(detail, indent=2))


# ---- polling: each check runs as a step; between checks the function is suspended ---

def arrived(state: dict, check) -> dict:
    listed = s3.list_objects_v2(Bucket=state["bucket"], Prefix=state["prefix"]).get("Contents", [])
    have = {o["Key"] for o in listed}
    return {**state, "missing": [k for k in state["expected"] if k not in have]}


def every_10_min_for_8_hours(state: dict, attempt: int) -> WaitForConditionDecision:
    if not state["missing"] or attempt >= 48:
        return WaitForConditionDecision.stop_polling()
    return WaitForConditionDecision.continue_waiting(Duration.from_minutes(10))


def glue_status(state: dict, check) -> dict:
    run = glue.get_job_run(JobName=GLUE_JOB, RunId=state["run_id"])["JobRun"]
    return {**state, "status": run["JobRunState"]}


def until_glue_finishes(state: dict, attempt: int) -> WaitForConditionDecision:
    if state["status"] in GLUE_DONE:
        return WaitForConditionDecision.stop_polling()
    return WaitForConditionDecision.continue_waiting(
        Duration.from_seconds(min(30 * 2 ** (attempt - 1), 600)))   # 30 s, 1 min, 2 min ... 10 min


# Starting a Glue run is a side effect with no idempotency token: run it at most once
# per attempt, and retry only when Glue says too many runs are already going.
START_ONCE = StepConfig(
    step_semantics=StepSemantics.AT_MOST_ONCE_PER_RETRY,
    retry_strategy=create_retry_strategy(RetryStrategyConfig(
        max_attempts=6, initial_delay=Duration.from_minutes(1), max_delay=Duration.from_minutes(10),
        retryable_error_types=[glue.exceptions.ConcurrentRunsExceededException])))


def check_file(ctx: DurableContext, item: dict, index: int, items: list) -> list[str]:
    return ctx.step(validate(item["bucket"], item["key"]), name=f"validate-{index}")


@durable_execution
def handler(job: dict, context: DurableContext) -> dict:
    batch = context.step(read_manifest(job["bucket"], job["key"]))

    # 1. Wait, without compute running, until every listed file has landed or 8 hours pass.
    batch = context.wait_for_condition(
        check=arrived, name="wait-for-files",
        config=WaitForConditionConfig(wait_strategy=every_10_min_for_8_hours,
                                      initial_state={**batch, "missing": batch["expected"]}))

    # 2. Validate what arrived, ten files at a time.
    present = [{"bucket": batch["bucket"], "key": k}
               for k in batch["expected"] if k not in batch["missing"]]
    checks = context.map(present, check_file, name="validate", config=MapConfig(max_concurrency=10))
    checks.throw_if_error()
    problems = ([p for found in checks.get_results() for p in found]
                + [f"missing: {k}" for k in batch["missing"]])

    # 3. Anything wrong: hold the batch and ask its owner. Nothing runs while they decide.
    if problems:
        def ask_owner(callback_id: str, wait_ctx) -> None:
            notify(f"Held: {batch['prefix']}", {"callback_id": callback_id, "problems": problems})
        reply = context.wait_for_callback(
            submitter=ask_owner, name="owner-decision",
            config=WaitForCallbackConfig(timeout=Duration.from_days(3)))
        decision = (json.loads(reply) if isinstance(reply, str) else reply)["decision"]
        if decision != "load":
            context.step(lambda _: notify(f"Rejected: {batch['prefix']}", problems), name="notify-rejected")
            return {"status": "rejected", "problems": problems}

    # 4. Load it, then wait for the Glue run the same way.
    run_id = context.step(start_glue(batch), config=START_ONCE)
    run = context.wait_for_condition(
        check=glue_status, name="wait-for-glue",
        config=WaitForConditionConfig(wait_strategy=until_glue_finishes,
                                      initial_state={"run_id": run_id, "status": "STARTING"}))
    if run["status"] != "SUCCEEDED":
        context.step(lambda _: notify(f"Load failed: {batch['prefix']}", run), name="notify-failed")
        raise RuntimeError(f"Glue run {run_id} ended {run['status']}")

    # 5. Record the load.
    context.step(commit(batch, run_id))
    return {"status": "loaded", "glue_run_id": run_id, "files": len(present)}

# The data owner, or a small approval tool, releases a held batch:
aws lambda send-durable-execution-callback-success \
  --callback-id "$CALLBACK_ID" \
  --cli-binary-format raw-in-base64-out \
  --result '{"decision": "load"}'

# Publish an immutable version and move the alias to it (create-alias the first time).
# Paused executions resume on the version they started on; new ones use the alias.
VERSION=$(aws lambda publish-version --function-name ingest-batch --query Version --output text)
aws lambda update-alias --function-name ingest-batch --name prod --function-version "$VERSION"

# A durable execution that fails is not retried after an asynchronous start.
# Send the triggering event to a dead-letter queue so it can be replayed.
aws lambda update-function-configuration --function-name ingest-batch \
  --dead-letter-config TargetArn=arn:aws:sqs:us-east-1:123456789012:ingest-batch-dlq

# backfill: a durable function that reloads a list of past batches, four at a time.
# Each batch runs as its own child execution with its own operation budget.
import os
from aws_durable_execution_sdk_python import DurableContext, durable_execution
from aws_durable_execution_sdk_python.config import CompletionConfig, MapConfig

INGEST_ARN = os.environ["INGEST_ARN"]     # qualified: ...:function:ingest-batch:prod

@durable_execution
def handler(event: dict, context: DurableContext) -> dict:
    def load(ctx: DurableContext, manifest_key: str, index: int, keys: list) -> dict:
        return ctx.invoke(INGEST_ARN, {"bucket": event["bucket"], "key": manifest_key},
                          name=f"batch-{index}")

    result = context.map(
        event["manifests"], load, name="backfill",
        config=MapConfig(max_concurrency=4,
                         completion_config=CompletionConfig(tolerated_failure_count=3)))
    return {"loaded": result.success_count, "failed": result.failure_count}

No. They're a Lambda feature, launched in December 2025. S3 is where they fit naturally in data platforms: files land in S3, S3 events start the executions, and since October 2026 the Python and JavaScript SDKs can keep state larger than 1 MB on an Amazon S3 Files or EFS mount.

Yes. S3 event notifications and EventBridge invoke Lambda asynchronously, so the execution can run for up to a year, as long as the target is a version or an alias. But they can't pass an execution name, so duplicate events start duplicate executions. A small dispatcher function that names the execution after the S3 event makes the start idempotent.

Lambda invokes it again and replays the checkpoint log. Steps that finished return their saved results without running again, and execution continues from the first step that hadn't finished. That step may run a second time, so steps should be idempotent, or use at-most-once semantics when they can't be.

Not for compute. During waits, callbacks and retry backoff the function is suspended and bills no duration. You pay per durable operation, which includes each polling check, plus a small charge for checkpoint data written and retained.

Not by default. Step Functions remains a strong choice for workflows that mainly connect AWS services, such as running a Glue job and waiting for it through a native integration. Durable functions suit new per-file and per-batch logic that waits on partners or people, written and tested as ordinary code.

Yes, but behind an event source mapping the whole execution must finish within 15 minutes, or 90 on Lambda Managed Instances. For longer workflows, let a plain function receive the messages and start the durable function asynchronously with an execution name.