Outbox pattern with DynamoDB Streams and EventBridge Pipes

The outbox pattern with DynamoDB solves a problem I deliberately left open in part 1 of this series. There, the intake function first writes a job to DynamoDB and then sends a message to SQS. Most of the time that works. If the function dies exactly between the two calls, though, a job sits in PENDING forever and nobody ever processes it. Gaps like this never show up in tests, but sooner or later they will show up in production.

In this part I close that gap without writing a single line of glue code for forwarding. The intake function only writes to the table. An EventBridge pipe turns the table’s change stream into the message for the worker, and a second pipe announces finished reports as a domain event on an event bus. The example is once again the fictional bookshop with its stock reports, this time built and tested entirely with the AWS CDK.

In short

With the outbox pattern, the application only writes to its database, and a separate process reliably turns those writes into messages. DynamoDB Streams and EventBridge Pipes give you that process ready-made: filtering, reshaping, retries and a dead-letter queue are configuration instead of code. The biggest pitfalls are details such as the stream’s starting position and filters that have to match the DynamoDB JSON format.

What you will learn in this article

• Why two writes in a row are never completely safe
• How DynamoDB Streams and EventBridge Pipes form an outbox together
• How to build pipes in the CDK with filter, reshaping and DLQ
• When you need explicit outbox entries written in a transaction
• Which settings really matter in production

Why use the outbox pattern with DynamoDB?

Writing to two systems one after the other, without both writes succeeding or failing together, is known as the dual-write problem. In part 1 it affects DynamoDB and SQS. There is no API for a shared transaction across both services. You could swap the order and send first, then store, but then a failure leaves a message in the queue that has no matching job. One of the two sides can always end up dangling.

The code from part 1 at least handles the most common case: if SQS answers with an error, the function marks the job as FAILED. That does not help against a timeout or a runtime crash between the two calls. With a bit of bad luck, exactly that happens during a traffic spike, when everyone is already staring at the dashboards.

Left: part 1 with two writes. Right: the outbox pattern with only one.

The idea behind the outbox pattern

This pattern comes from the world of relational databases and is described on microservices.io as the “transactional outbox”. The application stores its business change and the outgoing message in the same local transaction, with the message going into an outbox table. A separate process, the relay, reads that table and publishes the messages. That leaves only one write that matters. Either everything is stored or nothing is.

The price is an honest promise: the relay delivers at least once, not exactly once. If it crashes after sending but before recording its progress, it sends the same message again. Your consumers therefore have to be idempotent. I covered how to do that in detail in idempotency in AWS Lambda. The worker from part 1 already is, because claimJob rejects finished jobs with a condition.

Why DynamoDB Streams already include the outbox

A DynamoDB stream is a time-ordered log of every change to a table. According to the developer guide, every record appears exactly once in the stream, and the changes to an item arrive in the order in which they happened. The stream keeps the data for 24 hours. At its core that is exactly the outbox, except that DynamoDB maintains it for you.

For the job pattern this means: the new job itself is the event. As soon as the item with PENDING is in the table, it is also in the stream. In this case you do not need a separate outbox table. All that is missing is the relay that turns the stream record into an SQS message.

The relay: EventBridge Pipes instead of your own Lambda

In the past I would have attached a small Lambda function to the stream: read the record, check whether it is relevant, send the message. That works, but it is code you have to test, deploy and monitor. EventBridge Pipes do exactly this job as a managed service. A pipe connects a source with a target and can filter, enrich and reshape along the way.

How a pipe reads the stream

According to the Pipes documentation, a pipe polls the shards of a DynamoDB stream four times per second by default and invokes the target as soon as records are available. The order per item is preserved. The pipe retries failed records according to a strategy you define and then moves them to its own dead-letter queue. To decide whether EventBridge is the right service in the first place, the AWS service finder helps.

🎬 Flow of the outbox pattern with DynamoDB Streams and EventBridge Pipes – this animated diagram only runs on my site: watch it on sascha-ms.de

Two pipes for two questions

In the example, two pipes are attached to the same stream. The first answers the question “Was a new job created?” and then sends the job ID to the jobs queue. The second asks “Has a report just been finished?” and then publishes the ReportReady event on a dedicated event bus. Anyone interested in finished reports later, such as a mail sender or a statistics service, subscribes to that bus and never needs to know the jobs table. I explained why such business-named events matter so much to me in event-driven thinking.

Two is not a random number. The developer guide recommends that no more than two processes read from the same shard at the same time, otherwise throttling may occur. With two pipes the budget is used up. If you need more consumers, distribute through the event bus rather than adding pipes to the stream. That is exactly what part 3 is about.

Building the outbox pattern with DynamoDB in the CDK

Part 1 ran on the Serverless Framework, this time I use the AWS CDK. For pipes, aws-cdk-lib currently only offers the L1 construct CfnPipe, a direct mapping of the CloudFormation resource. The L2 constructs are still alpha packages. I stick with CfnPipe because the properties then map one to one to the documentation and no alpha version ends up in the project.

Prerequisites

• Node.js 22.18 or newer (for tests and the CDK app without a build step), Node.js 24 is recommended
• An AWS account with permissions for Lambda, API Gateway, DynamoDB, SQS, EventBridge, IAM and CloudFormation, plus the AWS CLI with valid credentials
• An account prepared once with npx cdk bootstrap in the target region
• The files from part 1. If the CDK is new to you, start with deploying a Lambda function with CDK and TypeScript.

Step 1: Create the project

Create a new folder and install the packages. What is missing stands out: the application no longer needs @aws-sdk/client-sqs, because no function sends to SQS itself.

mkdir bookshop-reports-outbox && cd bookshop-reports-outbox npm init -y npm install aws-cdk-lib constructs @aws-sdk/client-dynamodb @aws-sdk/lib-dynamodb npm install --save-dev aws-cdk typescript esbuild @types/node @types/aws-lambda aws-sdk-client-mock mkdir bin lib src test 

Then replace the package.json. "type": "module" makes Node.js run the TypeScript files as ES modules. You need esbuild locally so the CDK can bundle the Lambda functions without Docker.

{ "name": "bookshop-reports-outbox", "version": "1.0.0", "private": true, "type": "module", "scripts": { "typecheck": "tsc --noEmit", "test": "node --test test/*.test.ts", "synth": "cdk synth" }, "dependencies": { "@aws-sdk/client-dynamodb": "^3.1156.0", "@aws-sdk/lib-dynamodb": "^3.1156.0", "aws-cdk-lib": "^2.274.0", "constructs": "^10.8.1" }, "devDependencies": { "@types/aws-lambda": "^8.10.164", "@types/node": "^24.0.0", "aws-cdk": "^2.1146.1", "aws-sdk-client-mock": "^4.1.0", "esbuild": "^0.28.2", "typescript": "^7.0.2" } } 

{ "compilerOptions": { "target": "ES2023", "module": "nodenext", "moduleResolution": "nodenext", "strict": true, "noEmit": true, "allowImportingTsExtensions": true, "erasableSyntaxOnly": true, "verbatimModuleSyntax": true, "skipLibCheck": true, "types": ["node"] }, "include": ["bin", "lib", "src", "test"] } 

The cdk.json starts the app directly with Node.js. Since version 22.18, Node.js strips the type annotations itself, so a detour through ts-node is no longer needed:

{ "app": "node bin/app.ts" } 

// bin/app.ts import { App } from 'aws-cdk-lib'; import { ReportsStack } from '../lib/reports-stack.ts'; const app = new App(); new ReportsStack(app, 'BookshopReportsOutbox', { env: { region: 'eu-central-1' }, }); 

Next, copy the files src/jobs.ts, src/http.ts, src/get-report.ts, src/build-report.ts and src/report-worker.ts from part 1 into the src folder unchanged. The worker still reads a JSON object with jobId from the message. Where that message comes from does not matter to it.

Step 2: The stack with table, queues and pipes

This is where the actual pattern lives. Read the file once in full, I explain the important parts right after it.

// lib/reports-stack.ts import { CfnOutput, Duration, RemovalPolicy, Stack, type StackProps } from 'aws-cdk-lib'; import type { Construct } from 'constructs'; import { HttpApi, HttpMethod } from 'aws-cdk-lib/aws-apigatewayv2'; import { HttpLambdaIntegration } from 'aws-cdk-lib/aws-apigatewayv2-integrations'; import { AttributeType, BillingMode, StreamViewType, Table } from 'aws-cdk-lib/aws-dynamodb'; import { EventBus, Rule } from 'aws-cdk-lib/aws-events'; import { CloudWatchLogGroup } from 'aws-cdk-lib/aws-events-targets'; import { Role, ServicePrincipal } from 'aws-cdk-lib/aws-iam'; import { Architecture, Runtime } from 'aws-cdk-lib/aws-lambda'; import { SqsEventSource } from 'aws-cdk-lib/aws-lambda-event-sources'; import { NodejsFunction } from 'aws-cdk-lib/aws-lambda-nodejs'; import { LogGroup, RetentionDays } from 'aws-cdk-lib/aws-logs'; import { CfnPipe } from 'aws-cdk-lib/aws-pipes'; import { Queue } from 'aws-cdk-lib/aws-sqs'; // New jobs: only items created with status PENDING become a message. export const NEW_JOB_PATTERN = { dynamodb: { NewImage: { status: { S: ['PENDING'] } } }, }; // Finished reports: exactly the transition RUNNING -> SUCCEEDED. export const REPORT_DONE_PATTERN = { dynamodb: { OldImage: { status: { S: ['RUNNING'] } }, NewImage: { status: { S: ['SUCCEEDED'] } }, }, }; export class ReportsStack extends Stack { constructor(scope: Construct, id: string, props?: StackProps) { super(scope, id, props); const table = new Table(this, 'JobsTable', { partitionKey: { name: 'jobId', type: AttributeType.STRING }, billingMode: BillingMode.PAY_PER_REQUEST, timeToLiveAttribute: 'expiresAt', stream: StreamViewType.NEW_AND_OLD_IMAGES, removalPolicy: RemovalPolicy.DESTROY, // for the example only }); const jobsDlq = new Queue(this, 'JobsDeadLetterQueue', { retentionPeriod: Duration.days(14), }); const jobsQueue = new Queue(this, 'JobsQueue', { visibilityTimeout: Duration.seconds(720), // 6 x worker timeout deadLetterQueue: { queue: jobsDlq, maxReceiveCount: 5 }, }); // Stream records the pipe could not deliver end up here. const pipeDlq = new Queue(this, 'PipeDeadLetterQueue', { retentionPeriod: Duration.days(14), }); // For trying it out, all bookshop events land in a log group. const bus = new EventBus(this, 'BookshopBus', { eventBusName: 'bookshop' }); const eventLog = new LogGroup(this, 'BookshopEvents', { retention: RetentionDays.ONE_WEEK, removalPolicy: RemovalPolicy.DESTROY, }); new Rule(this, 'BookshopEventsToLog', { eventBus: bus, eventPattern: { source: ['bookshop.reports'] }, targets: [new CloudWatchLogGroup(eventLog, { installLatestAwsSdk: false })], }); // The pipes use their own role. const pipeRole = new Role(this, 'PipeRole', { assumedBy: new ServicePrincipal('pipes.amazonaws.com'), }); table.grantStreamRead(pipeRole); jobsQueue.grantSendMessages(pipeRole); pipeDlq.grantSendMessages(pipeRole); bus.grantPutEventsTo(pipeRole); const streamSource = { startingPosition: 'TRIM_HORIZON', batchSize: 10, maximumRetryAttempts: 10, maximumRecordAgeInSeconds: 3600, onPartialBatchItemFailure: 'AUTOMATIC_BISECT', deadLetterConfig: { arn: pipeDlq.queueArn }, }; // Pipe 1: new job -> message in the jobs queue const jobsPipe = new CfnPipe(this, 'NewJobsPipe', { roleArn: pipeRole.roleArn, source: table.tableStreamArn!, sourceParameters: { dynamoDbStreamParameters: streamSource, filterCriteria: { filters: [{ pattern: JSON.stringify(NEW_JOB_PATTERN) }] }, }, target: jobsQueue.queueArn, targetParameters: { inputTemplate: '{"jobId": <$.dynamodb.NewImage.jobId.S>}', }, }); // Pipe 2: report finished -> domain event on the event bus const donePipe = new CfnPipe(this, 'ReportDonePipe', { roleArn: pipeRole.roleArn, source: table.tableStreamArn!, sourceParameters: { dynamoDbStreamParameters: streamSource, filterCriteria: { filters: [{ pattern: JSON.stringify(REPORT_DONE_PATTERN) }] }, }, target: bus.eventBusArn, targetParameters: { eventBridgeEventBusParameters: { source: 'bookshop.reports', detailType: 'ReportReady', }, inputTemplate: '{"jobId": <$.dynamodb.NewImage.jobId.S>, "branch": <$.dynamodb.NewImage.branch.S>, "finishedAt": <$.dynamodb.NewImage.updatedAt.S>}', }, }); // The pipes may only be created once the role has its permissions. jobsPipe.node.addDependency(pipeRole); donePipe.node.addDependency(pipeRole); const fnDefaults = { runtime: Runtime.NODEJS_24_X, architecture: Architecture.ARM_64, memorySize: 256, timeout: Duration.seconds(10), environment: { JOBS_TABLE: table.tableName }, }; const createReport = new NodejsFunction(this, 'CreateReport', { ...fnDefaults, entry: 'src/create-report.ts', }); table.grant(createReport, 'dynamodb:PutItem', 'dynamodb:GetItem'); const getReport = new NodejsFunction(this, 'GetReport', { ...fnDefaults, entry: 'src/get-report.ts', }); table.grant(getReport, 'dynamodb:GetItem'); const worker = new NodejsFunction(this, 'ReportWorker', { ...fnDefaults, entry: 'src/report-worker.ts', timeout: Duration.seconds(120), environment: { JOBS_TABLE: table.tableName, MAX_ATTEMPTS: '5', REPORT_DELAY_MS: '20000' }, }); table.grant(worker, 'dynamodb:UpdateItem'); worker.addEventSource( new SqsEventSource(jobsQueue, { batchSize: 1, reportBatchItemFailures: true, maxConcurrency: 5, }), ); const api = new HttpApi(this, 'ReportsApi'); api.addRoutes({ path: '/reports', methods: [HttpMethod.POST], integration: new HttpLambdaIntegration('CreateReportIntegration', createReport), }); api.addRoutes({ path: '/reports/{jobId}', methods: [HttpMethod.GET], integration: new HttpLambdaIntegration('GetReportIntegration', getReport), }); new CfnOutput(this, 'ApiUrl', { value: api.apiEndpoint }); new CfnOutput(this, 'PipeDlqUrl', { value: pipeDlq.queueUrl }); new CfnOutput(this, 'EventLogGroup', { value: eventLog.logGroupName }); } } 

The table gets a stream with NEW_AND_OLD_IMAGES. Each record therefore contains the item before and after the change. The first pipe would only need the new image, but the second compares the old and new status. Choose the type carefully: according to the developer guide it cannot be changed later. For that you have to disable the stream and create a new one.

Step 3: Filters that match the DynamoDB format

A stream record contains the attributes in DynamoDB JSON format, that is, with type descriptors. A status does not appear as "status": "PENDING" but as "status": {"S": "PENDING"}. That is why the filter patterns look a little unusual:

{ "dynamodb": { "NewImage": ...