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": { "status": { "S": ["PENDING"] } }
  }
}

The obvious idea would be to filter on "eventName": ["INSERT"]. However, the Pipes filtering documentation lists eventName among the fields that cannot be used in filter patterns. So I filter on the content instead. For the job pattern that is enough: PENDING only appears on creation, because claimJob always sets the status to RUNNING.

The second pipe checks both images: RUNNING before, SUCCEEDED after. That way it fires exactly on the transition and not on every later change to a finished job. By the way, you do not have to clean up filtered records yourself: the pipe simply moves on in the stream.

Step 4: Shape the message

Without reshaping, the pipe would put the complete stream record into the queue. The worker, however, only expects {"jobId": "…"}. The input template builds exactly that shape:

{"jobId": <$.dynamodb.NewImage.jobId.S>}

Angle brackets contain a JSON path into the record. You do not need quotes around the value, for strings the pipe adds them itself. Watch out for the .S at the end: without it, the object {"S": "…"} ends up in the message and the worker finds no job ID. If a path is missing entirely, the documentation says the pipe inserts an empty value and reports no error. A typo in the template therefore only surfaces in the worker.

Step 5: Retries and dead-letter queue

The source parameters in streamSource decide how the pipe behaves on errors. The defaults are dangerous: according to the CloudFormation reference, MaximumRetryAttempts and MaximumRecordAgeInSeconds are set to -1, which means infinite. A single record that can never be delivered then blocks its shard until it drops out of the stream after 24 hours. Everything behind it waits.

Setting	Value in the example	Why
startingPosition	TRIM_HORIZON	Reads from the oldest record. With LATEST, changes made while the pipe is created or updated can be lost.
batchSize	10	SQS and EventBridge accept at most ten entries per batch.
maximumRetryAttempts	10	Limits the retries instead of retrying forever.
maximumRecordAgeInSeconds	3600	Discards records after one hour, well before the 24 hours run out.
onPartialBatchItemFailure	AUTOMATIC_BISECT	Splits a failing batch until the culprit record is isolated.
deadLetterConfig	dedicated SQS queue	Keeps records the pipe gave up on so you can investigate them.

The documentation does not describe in detail what exactly lands in the pipe DLQ. So I would not rely on finding the complete item there. With the outbox pattern that is not a big deal, because the job is still in the table and can be restarted from there. The error handling documentation mentions two subtleties: records older than maximumRecordAgeInSeconds go straight to the DLQ without a delivery attempt. And AWS retries internal pipe errors regardless of your limit until the record expires in the stream. Either way, the DLQ needs an alarm. Monitoring Lambda shows how to set that up with the CDK.

The CDK grants the pipe role its permissions through grantStreamRead, grantSendMessages and grantPutEventsTo. This results in exactly the four DynamoDB actions that the permissions page requires for a stream, plus sqs:SendMessage for queue and DLQ and events:PutEvents for the bus. The dependency via addDependency makes CloudFormation create the pipe only after the role’s policy exists.

Step 6: The intake function gets shorter

Compared to part 1, only the intake function changes. The SQS client, the error branch with 503 and the subsequent FAILED are gone:

// src/create-report.ts
import type { APIGatewayProxyEventV2, APIGatewayProxyStructuredResultV2 } from 'aws-lambda';
import { insertJob, jobIdFor, loadJob, newJob, type ReportJob } from './jobs.ts';
import { json } from './http.ts';

const BRANCH = /^[a-z0-9-]{2,40}$/;

function accepted(job: ReportJob): APIGatewayProxyStructuredResultV2 {
  const statusUrl = `/reports/${job.jobId}`;
  return json(
    202,
    { jobId: job.jobId, status: job.status, statusUrl },
    { location: statusUrl, 'retry-after': '5' },
  );
}

export async function handler(
  event: APIGatewayProxyEventV2,
): Promise<APIGatewayProxyStructuredResultV2> {
  let branch: unknown;
  try {
    branch = JSON.parse(event.body ?? '{}').branch;
  } catch {
    return json(400, { message: 'Body is not valid JSON' });
  }
  if (typeof branch !== 'string' || !BRANCH.test(branch)) {
    return json(400, { message: 'branch is missing or has an invalid format' });
  }

  const key = event.headers?.['idempotency-key'];
  if (key !== undefined && (key.length < 8 || key.length > 128)) {
    return json(400, { message: 'Idempotency-Key must be 8 to 128 characters long' });
  }
  const job = newJob(jobIdFor(key), branch);

  if (!(await insertJob(job))) {
    // Repeated call with the same key: return the existing job.
    const existing = await loadJob(job.jobId);
    if (!existing) return json(404, { message: 'Job has expired' });
    if (existing.branch !== branch) {
      return json(409, { message: 'Idempotency-Key was already used for a different request' });
    }
    return accepted(existing);
  }

  // No SendMessage any more: the pipe turns the new item into the message.
  return accepted(job);
}

The permissions shrink as well. The function may only run PutItem and GetItem on the table, and it does not know SQS at all. Fewer permissions mean less that can go wrong in case of a security hole.

Step 7: Test the stack

Without an AWS account you can still check whether the CloudFormation template contains the right things. The tests use the assertions from aws-cdk-lib and the Node.js test runner:

// test/stack.test.ts
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { App } from 'aws-cdk-lib';
import { Match, Template } from 'aws-cdk-lib/assertions';
import { ReportsStack, NEW_JOB_PATTERN, REPORT_DONE_PATTERN } from '../lib/reports-stack.ts';

const template = Template.fromStack(new ReportsStack(new App(), 'TestStack'));

test('table has a stream with old and new images', () => {
  template.hasResourceProperties('AWS::DynamoDB::Table', {
    StreamSpecification: { StreamViewType: 'NEW_AND_OLD_IMAGES' },
    TimeToLiveSpecification: { AttributeName: 'expiresAt', Enabled: true },
  });
});

test('new-jobs pipe filters on PENDING and builds the message', () => {
  template.hasResourceProperties('AWS::Pipes::Pipe', {
    SourceParameters: {
      FilterCriteria: { Filters: [{ Pattern: JSON.stringify(NEW_JOB_PATTERN) }] },
      DynamoDBStreamParameters: Match.objectLike({
        StartingPosition: 'TRIM_HORIZON',
        MaximumRetryAttempts: 10,
        DeadLetterConfig: Match.anyValue(),
      }),
    },
    TargetParameters: { InputTemplate: '{"jobId": <$.dynamodb.NewImage.jobId.S>}' },
  });
});

test('report-ready pipe sends ReportReady to the bus', () => {
  template.hasResourceProperties('AWS::Pipes::Pipe', {
    SourceParameters: Match.objectLike({
      FilterCriteria: { Filters: [{ Pattern: JSON.stringify(REPORT_DONE_PATTERN) }] },
    }),
    TargetParameters: Match.objectLike({
      EventBridgeEventBusParameters: { Source: 'bookshop.reports', DetailType: 'ReportReady' },
    }),
  });
});

test('pipe role may read the stream and write to the queues', () => {
  const policies = template.findResources('AWS::IAM::Policy');
  const actions = JSON.stringify(
    Object.entries(policies)
      .filter(([id]) => id.startsWith('PipeRole'))
      .map(([, p]) => p.Properties.PolicyDocument.Statement),
  );
  for (const a of ['dynamodb:GetRecords', 'dynamodb:GetShardIterator', 'dynamodb:DescribeStream', 'dynamodb:ListStreams', 'sqs:SendMessage', 'events:PutEvents']) {
    assert.ok(actions.includes(a), `${a} is missing`);
  }
});

test('intake function no longer needs SQS permissions', () => {
  const policies = template.findResources('AWS::IAM::Policy');
  const create = Object.entries(policies).find(([id]) => id.startsWith('CreateReport'));
  assert.ok(create);
  assert.ok(!JSON.stringify(create[1]).includes('sqs:'));
});

npm run typecheck
npm test
npx cdk synth --quiet

Expected result: tsc reports no errors, the test runner shows # pass 5 and # fail 0, and cdk synth writes the template to the cdk.out folder. On the first run, esbuild bundles the three functions, which takes a few seconds.

Step 8: Deploy and try it out

npx cdk deploy

At the end, the output shows ApiUrl, PipeDlqUrl and EventLogGroup. Give the pipes a few minutes afterwards. According to the documentation, stream polling during pipe creation is eventually consistent, and startup can take several minutes. Then send a request:

API=https://abc123.execute-api.eu-central-1.amazonaws.com
curl -i -X POST "$API/reports" \
  -H "content-type: application/json" \
  -H "idempotency-key: stock-hafen-2026-10" \
  -d '{"branch":"hafen"}'

The response looks like the one in part 1, except that the pipe now creates the message in the background. After about 25 seconds the status endpoint returns the result. Sample output, shortened:

HTTP/2 202
location: /reports/0d4f6a8c2b1e9f7a5c3d2e1f0a9b8c7d
retry-after: 5

{"jobId":"0d4f6a8c2b1e9f7a5c3d2e1f0a9b8c7d","status":"PENDING","statusUrl":"/reports/0d4f6a8c2b1e9f7a5c3d2e1f0a9b8c7d"}

The second channel is more interesting. Follow the log group from the output and you will see the event as soon as the report is finished:

aws logs tail <EventLogGroup> --follow --format short

{"version":"0","detail-type":"ReportReady","source":"bookshop.reports", … ,"detail":{"jobId":"0d4f6a8c2b1e9f7a5c3d2e1f0a9b8c7d","branch":"hafen","finishedAt":"2026-10-25T18:41:07.512Z"}}

The values above are samples. When you are done, npx cdk destroy removes everything again, including the table, because it carries RemovalPolicy.DESTROY in the example. For real data I would never set that.

Variant: explicit outbox entries in a transaction

With the job pattern, the item itself is the event. That does not always fit. Sometimes you want to publish an event that cannot be derived from a state change, such as “report requested” with details that are not stored in the job at all. Or a single change should trigger several different events. In that case you write an explicit outbox entry, and you do it in the same transaction as the business change.

DynamoDB offers TransactWriteItems for this. According to the developer guide, the call bundles up to 100 write actions that all succeed or fail together, with a total size of at most 4 MB. The same item may only appear once per transaction, so the job and the outbox entry need different keys:

// src/outbox.ts
import { TransactWriteCommand } from '@aws-sdk/lib-dynamodb';
import { randomUUID } from 'node:crypto';
import { db, type ReportJob } from './jobs.ts';

// An outbox entry lives in the same table as the jobs.
// The "outbox#" prefix does not pass the job ID check in the status endpoint.
export interface OutboxEntry {
  jobId: string;
  kind: 'OUTBOX';
  eventType: string;
  subjectId: string;
  branch: string;
  createdAt: string;
  expiresAt: number;
}

export function outboxEntry(eventType: string, job: ReportJob, now = new Date()): OutboxEntry {
  return {
    jobId: `outbox#${randomUUID()}`,
    kind: 'OUTBOX',
    eventType,
    subjectId: job.jobId,
    branch: job.branch,
    createdAt: now.toISOString(),
    // One day is enough: the pipe reads the entry within seconds.
    expiresAt: Math.floor(now.getTime() / 1000) + 86_400,
  };
}

// Write job and event together: either both or neither.
export async function insertJobWithEvent(job: ReportJob, eventType: string): Promise<boolean> {
  const TableName = process.env.JOBS_TABLE;
  try {
    await db.send(
      new TransactWriteCommand({
        TransactItems: [
          { Put: { TableName, Item: job, ConditionExpression: 'attribute_not_exists(jobId)' } },
          { Put: { TableName, Item: outboxEntry(eventType, job) } },
        ],
      }),
    );
    return true;
  } catch (err) {
    const reasons = (err as { CancellationReasons?: { Code?: string }[] }).CancellationReasons;
    // Only the job already existed: that is the known idempotency case.
    if ((err as Error).name === 'TransactionCanceledException' && reasons?.[0]?.Code === 'ConditionalCheckFailed') {
      return false;
    }
    throw err;
  }
}

Handling a cancelled transaction

If the transaction is cancelled, DynamoDB returns a TransactionCanceledException with a reason per action. Only if the first reason is ConditionalCheckFailed did the job already exist. A conflict with a parallel transaction, on the other hand, is a real error and gets rethrown. Thanks to its short expiry, TTL removes the outbox entry after one day. The deletion does create another stream record, but that record has no new image and therefore fails every filter.

A pipe for the outbox entries

The matching pipe filters on the kind attribute and builds the event from the flat attributes:

// Addition to lib/reports-stack.ts
new CfnPipe(this, 'OutboxPipe', {
  roleArn: pipeRole.roleArn,
  source: table.tableStreamArn!,
  sourceParameters: {
    dynamoDbStreamParameters: streamSource,
    filterCriteria: {
      filters: [{ pattern: JSON.stringify({ dynamodb: { NewImage: { kind: { S: ['OUTBOX'] } } } }) }],
    },
  },
  target: bus.eventBusArn,
  targetParameters: {
    eventBridgeEventBusParameters: { source: 'bookshop.reports', detailType: 'ReportRequested' },
    inputTemplate:
      '{"eventType": <$.dynamodb.NewImage.eventType.S>, "jobId": <$.dynamodb.NewImage.subjectId.S>, "branch": <$.dynamodb.NewImage.branch.S>}',
  },
});

Keep two things in mind. First, this would be the third pipe on the same stream and thus above the recommendation of two readers per shard. In practice I would therefore drop the second pipe and write ReportReady as an outbox entry too. Second, the pipe cannot unpack DynamoDB JSON. If you store the content as a nested map, the type descriptors end up in the event. Keep the fields flat or put a Lambda enrichment in between that converts the item.

Common mistakes with the outbox pattern in DynamoDB

Disabling and re-enabling the stream

It sounds harmless, but it is not. According to the Pipes documentation, a pipe becomes unusable as soon as its source stream is disabled, even if you enable it again afterwards. The new stream gets a new ARN, and you have to create the pipe again. In the CDK this happens when you change the table’s stream type. So plan the type right from the start.

Changing the starting position later

In CloudFormation, startingPosition is also a property whose change replaces the pipe. Also consider what TRIM_HORIZON means for a new pipe: it starts at the oldest record, up to 24 hours in the past. If CloudFormation replaces a pipe, for example because you change the construct ID or the starting position, the messages and events from that period arrive once more. The idempotent worker copes with that, and consumers of ReportReady have to as well. With LATEST it would be the other way round: no duplicates, but possible gaps.

More than two readers per shard

Every pipe and every Lambda function attached directly to the stream counts as a reader. From the third one on, throttling becomes likely and latency rises. Collect events on an event bus instead and distribute them there with rules. If you need more than 24 hours of history or many readers, take a look at Kinesis Data Streams for DynamoDB.

Consumers that are not idempotent

Pipes deliver at least once. Retries after a partial failure and the replacement of a pipe lead to duplicate messages. The worker from part 1 handles that, a mail sender without protection would write twice. So give your events a stable ID, the jobId in the example, and check it on the consumer side. You can find more on errors in SQS consumers in SQS partial batch response.

What does the outbox pattern cost?

Pipes charge per request, and only after filtering. The AWS pricing example uses 0.40 US dollars per million requests, with every started 64 KB chunk of a payload counting as one request. Records dropped by the filter do not appear on the bill. For the second pipe this means: only the finished reports cost money, not every status change.

On top of that come the reads from the stream. The DynamoDB pricing page explicitly states that reads through Lambda triggers are free, but it says nothing about pipes. So to be safe, plan for stream read requests, which cost 0.02 US dollars per 100,000 requests after the free tier. If you use the transaction variant, every write in the transaction costs double, because DynamoDB prepares and then commits. On the other hand, you save the SQS call and a bit of runtime in the intake function.

My take

I like the outbox pattern with DynamoDB because it takes an uncomfortable truth seriously: distributed systems fail at the transitions, not in the middle. With streams and pipes, a pattern that needs its own relay process in the relational world turns into a few lines of configuration. The intake function becomes simpler, needs fewer permissions, and an event like ReportReady comes almost for free.

It is not entirely free, though. Logic moves out of the code into filter patterns and templates that no compiler checks and that are less visible when debugging than a console.log. That is why the patterns belong in the code as constants and in tests, as in the example. Also, not every application needs this guarantee. For an internal report that someone can simply restart if needed, the cleanup job from part 1 is perfectly fine. As soon as money, orders or customer communication depend on a message, however, I would build in the outbox pattern from day one.

FAQ

What is the outbox pattern?

A pattern in which an application stores its change and the outgoing message together in the database. A separate process reads these entries and publishes them. That way no message belonging to a stored change can get lost.

Do I need a separate table for the outbox pattern with DynamoDB?

Not necessarily. If the changed item itself is the event, the table’s DynamoDB stream is enough. Explicit outbox entries pay off when the event contains different data than the item. They can live in the same table and be written alongside with TransactWriteItems.

EventBridge Pipes or Lambda on the DynamoDB stream?

Pipes are the better choice when you only want to filter, reshape and forward. You need a Lambda function when real logic is required along the way, such as calling another service. It can also be added to a pipe as an enrichment step.

Does the outbox pattern guarantee exactly-once delivery?

No. It guarantees that no message gets lost. Duplicate deliveries are possible, for example after retries. Consumers therefore have to be idempotent, for instance through a stable ID and a condition on write.

Your next step

Pick a function in your project that first writes to a database and then sends a message. Ask yourself what happens if it crashes exactly in between. If the answer is “then something is missing”, it is a good candidate for the outbox pattern. Before that, deploy the example and play with the filter: change it so that FAILED jobs also produce an event.

Where do such dual writes hide in your applications, and how do you deal with them today? Let me know in the comments.

Next in the series: Overview “Serverless patterns in practice” · Previous part: Asynchronous API with Lambda · Next part: fan-out with SNS and SQS

Sources

• microservices.io: Pattern “Transactional outbox”
• Amazon DynamoDB: Change data capture for DynamoDB Streams
• Amazon EventBridge: DynamoDB Streams as a source for Pipes
• Amazon EventBridge: Event filtering in Pipes
• Amazon EventBridge: Input transformation in Pipes
• Amazon EventBridge: Batching and concurrency in Pipes
• Amazon EventBridge: Permissions for Pipes
• Amazon EventBridge: Error handling in Pipes
• AWS CloudFormation: PipeSourceDynamoDBStreamParameters
• Amazon DynamoDB: Transactions (TransactWriteItems)
• Amazon EventBridge: Pricing (Pipes)
• Amazon DynamoDB: On-demand pricing (Streams)
• Tested with Node.js 22.23, TypeScript 7.0.2, aws-cdk-lib 2.274.0, aws-cdk 2.1146.1, @aws-sdk/* 3.1156.0, esbuild 0.28.2: tsc, cdk synth, the 5 stack tests from the article and 6 additional mock tests for the intake function and the outbox variant. Deploy and log outputs are marked as samples.

This post was created with AI assistance.

---

👉 Next step: Need a schedule in EventBridge? My EventBridge cron helper builds and explains the expression for you.

Originally published on sascha-ms.de. More articles on serverless, AI in everyday development and leading tech teams: sascha-ms.de/en