S3 to Step Functions Event Flow in AWS Intelligent Document Processing: Complete EventBridge Pipeline Guide

The event flow begins when an S3 object creation triggers an EventBridge rule that invokes the QueueSender Lambda, which persists document metadata to DynamoDB and enqueues a message to SQS, finally triggering the QueueProcessor Lambda to start the Step Functions execution after a concurrency check.

The accelerated intelligent document processing (IDP) solution on AWS implements a fully serverless, event-driven pipeline that automatically initiates document processing workflows the moment a file lands in S3. Understanding the precise event flow from S3 object creation through EventBridge to Step Functions execution is essential for debugging, monitoring, and extending this accelerator.

Architecture Overview: Event-Driven Document Ingestion

The pipeline uses native AWS event routing without polling. When a document is uploaded to the InputBucket, S3 emits an event that flows through EventBridge, two Lambda functions, and SQS before finally launching the Step Functions state machine that orchestrates the specific IDP pattern (OCR, classification, or extraction).

Step-by-Step Event Flow Analysis

S3 Object Creation and EventBridge Enablement

The ingestion flow originates in template.yaml where the InputBucket resource explicitly enables EventBridge notifications:

EventBridgeConfiguration:
  EventBridgeEnabled: true

This configuration (lines 1832–1835 in template.yaml) instructs S3 to emit CloudWatch Events of type "Object Created" whenever a PUT or POST operation creates an object. Unlike traditional S3 event notifications, EventBridge enables advanced filtering and routing through the EventBridge bus.

EventBridge Rule Routing to QueueSender

An EventBridge rule defined within the QueueSender Lambda resource matches the S3 event and triggers the function. In template.yaml (lines 3636–3649), the Events section defines an S3Event rule:

Events:
  S3Event:
    Type: EventBridgeRule
    Properties:
      EventBusName: default
      Pattern:
        source:
          - aws.s3
        detail-type:
          - Object Created
        detail:
          bucket:
            name:
              - !Ref InputBucket

This rule filters for events from aws.s3 with detail-type "Object Created" specifically from the InputBucket, ensuring the QueueSender Lambda only processes relevant documents.

Document Record Creation and Persistence

The QueueSender Lambda (src/lambda/queue_sender/index.py) handles the EventBridge event. The handler function (lines 30–45) extracts the S3 key from detail.object.key and creates a Document object:

document = Document.from_s3_event(event, output_bucket)
document.status = "QUEUED"
document.trace_id = str(uuid.uuid4())

The Lambda then persists this metadata to DynamoDB via document_service.create_document, storing the document ID, input key, status, configuration version, and trace ID for end-to-end tracking.

SQS Message Queuing

After persistence, the QueueSender serializes the Document object and sends it to the DocumentQueue SQS queue. In src/lambda/queue_sender/index.py (lines 94–112):

sqs.send_message(
    QueueUrl=queue_url,
    MessageBody=document.to_json(),
    MessageAttributes={
        'EventType': {'StringValue': 'DocumentQueued', 'DataType': 'String'}
    }
)

This decouples the ingestion from processing, allowing the system to handle burst traffic and providing durable message storage if downstream components are temporarily unavailable.

QueueProcessor Invocation and Concurrency Control

The QueueProcessor Lambda (src/lambda/queue_processor/index.py) consumes messages from the DocumentQueue via a standard SQS trigger. The process_message function (lines 34–42) deserializes the payload:

document = Document.load_document(record['body'])

Before starting the workflow, the Lambda checks a concurrency control DynamoDB table to ensure the system does not exceed configured processing limits. This prevents throttling of downstream services like Textract or Comprehend.

Step Functions Execution Launch

If the concurrency check passes, the start_workflow function (lines 70–108) prepares the execution payload. It compresses the document object (or falls back to the raw dictionary if compression fails) and constructs the Step Functions input:

event = {
    'document': compressed_document,
    'execution_id': document.id
}

response = sfn.start_execution(
    stateMachineArn=state_machine_arn,
    input=json.dumps(event)
)

The state_machine_arn corresponds to the specific IDP pattern (Pattern-1, Pattern-2, or Pattern-3) defined in patterns/pattern-*/template.yaml. The state machine then orchestrates the document processing workflow, invoking Lambda functions for OCR, classification, and data extraction.

Key Implementation Files

File Path	Purpose
template.yaml (lines 1832–1835)	Enables S3 EventBridge notifications on InputBucket
template.yaml (lines 3636–3649)	Defines EventBridge rule routing S3 events to QueueSender
src/lambda/queue_sender/index.py	Processes S3 events, creates Document records, enqueues to SQS
src/lambda/queue_processor/index.py	Consumes SQS messages, manages concurrency, starts Step Functions
patterns/pattern-*/template.yaml	Defines pattern-specific Step Functions state machines

Practical Verification Examples

Upload a document to trigger the flow:

aws s3 cp ./samples/lending_package.pdf s3://my-idp-input-bucket/

Inspect the message queued by QueueSender:

aws sqs receive-message \
    --queue-url https://sqs.<region>.amazonaws.com/<account>/DocumentQueue \
    --max-number-of-messages 1 \
    --message-attribute-names All

Manually start a Step Functions execution for testing:

import boto3
import json

sfn = boto3.client('stepfunctions')

execution = sfn.start_execution(
    stateMachineArn='arn:aws:states:<region>:<account>:stateMachine:Pattern1StateMachine',
    input=json.dumps({
        "document": {
            "id": "doc-123",
            "input_key": "test.pdf",
            "status": "QUEUED"
        },
        "execution_id": "doc-123"
    })
)

print(f"Execution ARN: {execution['executionArn']}")

Summary

• S3 EventBridge Integration: The InputBucket enables native EventBridge notifications to emit "Object Created" events without custom polling.
• EventBridge Routing: A CloudWatch Events rule filters S3 events by bucket name and detail-type, triggering the QueueSender Lambda.
• Document Tracking: QueueSender creates a Document record in DynamoDB with status QUEUED and trace metadata before enqueueing.
• Decoupled Processing: Messages flow through SQS (DocumentQueue) to buffer traffic and ensure durability between ingestion and processing.
• Concurrency Control: QueueProcessor checks DynamoDB limits before invoking Step Functions to prevent downstream service throttling.
• Step Functions Orchestration: The state machine ARN is dynamically selected based on the IDP pattern, launching the document processing workflow.

Frequently Asked Questions

How does S3 notify EventBridge when a new object is uploaded?

The S3 bucket includes an EventBridgeConfiguration property with EventBridgeEnabled: true in the CloudFormation template (template.yaml lines 1832–1835). This enables S3 to emit CloudWatch Events of type "Object Created" directly to the default EventBridge bus whenever objects are created via PUT or POST operations, eliminating the need for S3 event notifications or polling mechanisms.

What is the purpose of the QueueSender Lambda function?

The QueueSender Lambda (src/lambda/queue_sender/index.py) acts as the ingestion adapter that transforms S3 events into trackable document records. It extracts the object key from the EventBridge payload, creates a Document object with status QUEUED, persists metadata to DynamoDB tables for configuration and tracking, and sends the serialized document to the DocumentQueue SQS queue with a DocumentQueued event type attribute.

How does the system prevent Step Functions execution overload?

The QueueProcessor Lambda implements concurrency control by checking a dedicated DynamoDB table before starting workflows (src/lambda/queue_processor/index.py). When processing messages from the SQS queue, it verifies that the current number of running executions is below the configured limit. Only if the concurrency check passes does it call sfn.start_execution with the compressed document payload and the appropriate state machine ARN, preventing throttling of downstream services like Amazon Textract or Comprehend.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}