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

> Discover how Lambda functions queue_sender and queue_processor manage queue processing and concurrency in AWS Accelerated IDP. Learn about DynamoDB-backed counter for limits.

- Repository: [aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws](https://github.com/aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws)
- Tags: internals
- Published: 2026-02-25

---

**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`](https://github.com/aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws/blob/main/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`](https://github.com/aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws/blob/main/src/lambda/queue_sender/index.py):

```python
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`](https://github.com/aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws/blob/main/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:

```python
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:

```python
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`](https://github.com/aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws/blob/main/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`](https://github.com/aws-solutions-library-samples/accelerated-intelligent-document-processing-on-aws/blob/main/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']`.