aws-samples / aws-samples/sample-automatic-sync-for-bedrock-knowledge-bases

Tracking table "processed" / "ingestion_job_id" never updated after successful ingestion (change_id not propagated through SQS)

Open
#5 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
4
Forks
3
PR merge metrics
No merged PRs in 30d

Description

## Summary

When the Event Processor Lambda writes a change record to the tracking DynamoDB table, the generated `change_id` is never included in the downstream SQS notification. The Monitor Sync Lambda expects `message.change_ids` to identify which tracking rows to mark as processed after an ingestion job completes, but with no IDs in the message it has nothing to update. As a result, every tracking row stays at `processed = false` with `ingestion_job_id` unset, even when ingestion completes successfully.

## Affected files

- `src/event_processor_lambda.py` (SQS message producer side)
- `src/monitor_sync_lambda.py` (consumer that should mark rows processed)

## Environment

- Default deployment from this repository's `main` (no modifications to the SQS plumbing).
- Tracking DynamoDB table enabled (default in the provided template).

## Steps to reproduce

1. Deploy the stack with the default configuration.
2. Upload an object under the watched prefix, e.g.
`aws s3 cp sample.pdf s3:////`
3. Wait for the ingestion job triggered by Step Functions to reach `COMPLETE`.
4. Inspect the tracking DynamoDB row created for this object.

Expected: the row has `processed = true` and `ingestion_job_id = `.

Actual: the row stays at `processed = false` and `ingestion_job_id` is not set.

## Root cause

In `event_processor_lambda.py`, the lambda generates a `change_id` for the tracking table but the SQS notification message does **not** carry that ID:

```python
change_id = str(uuid.uuid4())
tracking_table.put_item(Item={'change_id': change_id, ...})
...
message = {
'change_type': change_type,
'bucket': bucket,
'key': key,
...
}
# change_id never reaches `message`
```

In `monitor_sync_lambda.py`, the lambda reads `change_ids` from the message:

```python
change_ids = event.get('message', {}).get('change_ids', [])
```

and passes them to `mark_changes_as_processed(job_id, change_ids)`. Because the producer never sends `change_ids` (or `change_id`), this is always `[]`, and no tracking rows are updated.

## Proposed fix

1. In `event_processor_lambda.py`, only after a successful `tracking_table.put_item`, include `change_id` in the SQS message payload.
2. In `monitor_sync_lambda.py`, accept both `change_id` (single string) and `change_ids` (list) when collecting IDs to mark as processed, deduplicating.

This preserves backward compatibility: messages without `change_id` (e.g. when the tracking table is disabled or the put fails) behave as before.

Reference implementation in my fork: revsystem/sample-automatic-sync-for-bedrock-knowledge-bases@fe19871 (with a small follow-up cleanup at @3bfc1f7).

If this matches your understanding, I'd be glad to submit a PR per CONTRIBUTING.md.

Contributor guide

Open the contributing guide

Research direction

Start with src/event_processor_lambda.py and src/monitor_sync_lambda.py, tracing the successful tracking_table.put_item call through the SQS message and the consumer's change ID collection. Verify that single and list IDs are deduplicated and passed to mark_changes_as_processed. Deploy the default stack or reproduce the S3 upload flow, then confirm the tracking row becomes processed with the completed ingestion job ID.

Written by the indexing model from the issue text.

Assessment

Tech stack
aws, python
Domain
backend, cloud, databases
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
74/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.