Lambda Functions for Queue Processing and Concurrency Management in AWS Accelerated IDP

The queue_sender and queue_processor Lambda functions in the accelerated-intelligent-document-processing-on-aws solution handle queue processing and concurrency management, with queue_processor enforcing limits via a DynamoDB-backed counter.

The accelerated-intelligent-document-processing-on-aws solution from the AWS Solutions Library implements a serverless document ingestion pipeline that requires careful coordination between queue consumers and resource limits. Two specific Lambda functions—queue_sender and queue_processor—manage the flow of documents through Amazon SQS and enforce strict concurrency controls to prevent downstream Step Functions executions from overwhelming backend resources.

The queue_sender Lambda Function

The queue_sender function acts as the ingestion entry point, triggered by S3 EventBridge events when new documents arrive in the input bucket. Located at src/lambda/queue_sender/index.py, this function performs the initial orchestration steps without implementing concurrency logic itself.

When invoked, the function:

1. Builds a Document object containing metadata about the incoming file
2. Records the queued timestamp and writes the document metadata to DynamoDB
3. Serializes the document and posts it to the main SQS DocumentQueue

By deferring concurrency management to downstream consumers, queue_sender focuses solely on reliable message delivery. The relevant implementation for sending messages appears in lines 94-102 of src/lambda/queue_sender/index.py:

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

The queue_processor Lambda Function

The queue_processor function, implemented in src/lambda/queue_processor/index.py, handles the critical tasks of queue consumption and concurrency enforcement. This function consumes batches of up to 10 messages from the DocumentQueue and manages a DynamoDB-backed counter to enforce the user-defined MAX_CONCURRENT limit (defaulting to 5).

Concurrency Control Mechanism

The concurrency management relies on a conditional update pattern against the CONCURRENCY_TABLE DynamoDB table. Before launching a Step Functions workflow, the function attempts to increment the active_count only if the current value is less than MAX_CONCURRENT.

The update_counter implementation (lines 45-57) demonstrates this pattern:

def update_counter(increment: bool = True) -> bool:
    # Build DynamoDB UpdateExpression

    update_args = {
        'Key': {'counter_id': COUNTER_ID},
        'UpdateExpression': 'ADD active_count :inc',
        'ExpressionAttributeValues': {
            ':inc': 1 if increment else -1,
            ':max': MAX_CONCURRENT
        },
        'ReturnValues': 'UPDATED_NEW'
    }
    # Only allow increment when under the limit

    if increment:
        update_args['ConditionExpression'] = 'active_count < :max'
    response = concurrency_table.update_item(**update_args)
    return True

Error Handling and Counter Decrement

If the workflow fails to start or encounters an error after the counter has been incremented, the function decrements the counter to release the capacity slot. This ensures that transient failures do not permanently reduce pipeline throughput.

The error handling logic (lines 77-81) shows the decrement operation:

except Exception as e:
    logger.error(f"Error processing {object_key}: {str(e)}")
    # Release the slot so other documents can proceed

    try:
        update_counter(increment=False)
    except Exception as counter_error:
        logger.error(f"Failed to decrement counter: {counter_error}")
    return False, message_id

Architectural Flow Summary

The interaction between these two Lambda functions creates a robust back-pressure mechanism:

1. Ingestion: queue_sender receives S3 events, persists metadata to DynamoDB, and publishes to SQS
2. Consumption: queue_processor polls the queue in batches
3. Throttling: Before processing each document, queue_processor checks the DynamoDB counter against MAX_CONCURRENT
4. Execution: Only when capacity is available does the function increment the counter and launch the Step Functions workflow
5. Cleanup: Successful completion or startup failures trigger counter decrements to free capacity

Summary

• queue_sender (src/lambda/queue_sender/index.py) handles document ingestion by receiving S3 events, creating DynamoDB records, and publishing messages to SQS without implementing concurrency logic.
• queue_processor (src/lambda/queue_processor/index.py) consumes SQS batches and enforces the MAX_CONCURRENT limit using a DynamoDB conditional update pattern, incrementing counters before workflow execution and decrementing them on failures.
• The concurrency table (CONCURRENCY_TABLE) stores the active_count that both functions reference to prevent Step Functions from exceeding configured capacity limits.

Frequently Asked Questions

How does the solution prevent Step Functions from being overwhelmed?

The queue_processor Lambda uses a DynamoDB counter stored in the CONCURRENCY_TABLE to track active workflow executions. Before starting a new Step Functions workflow, it attempts to increment this counter only if the current active_count is less than the MAX_CONCURRENT environment variable (default 5). If the limit is reached, the message remains in the queue for later processing.

What happens when the concurrency limit is reached?

When queue_processor attempts to increment the counter and the condition active_count < MAX_CONCURRENT fails, DynamoDB raises a ConditionalCheckFailedException. The Lambda logs that the concurrency limit has been reached and does not delete the SQS message, allowing it to become visible again for retry once capacity frees up or the visibility timeout expires.

How does queue_sender differ from queue_processor?

The queue_sender function acts as the pipeline entry point, triggered by S3 events to create document metadata in DynamoDB and publish messages to SQS. It contains no concurrency logic. The queue_processor function acts as the throttled consumer, reading from SQS and enforcing the MAX_CONCURRENT limit via DynamoDB before launching Step Functions workflows.

Where is the concurrency limit configured?

The concurrency limit is defined by the MAX_CONCURRENT environment variable passed to the queue_processor Lambda function. This value is typically set in the CloudFormation templates or infrastructure-as-code definitions that deploy the solution, with a default value of 5 concurrent executions. The value is read at runtime from os.environ['MAX_CONCURRENT'].

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