Event Sourcing auf AWS: DynamoDB Streams und Lambda

In der Konzept-Betrachtung zum Event Sourcing habe ich den Event-Store als Logbuch beschrieben: Der Zustand ist nicht das Primäre, sondern die Folge der Ereignisse, die zu ihm geführt haben. Das ist ein schönes Modell auf dem Whiteboard. In Kundenprojekten kommt danach die Frage, die alles entscheidet: Wo läuft das? Wer garantiert die Reihenfolge? Was passiert, wenn ein Consumer denselben Event zweimal sieht?

Ich baue diese Systeme inzwischen fast immer serverless auf AWS – und zwar mit einer überraschend kleinen Menge an Bausteinen. DynamoDB ist der Event-Store, DynamoDB Streams sind der Change-Feed, und Lambda ist der Projektor, der aus den Events die Read-Models der Leseseite faltet. In diesem Beitrag zeige ich, wie diese drei Teile zusammenspielen, wo die scharfen Kanten liegen und welchen Code ich dafür schreibe.

Die drei Bausteine

Bevor ich in den Code gehe, das mentale Modell. Wer die Rollen sauber trennt, macht später deutlich weniger Fehler:

Der Event-Store ist eine DynamoDB-Tabelle, append-only: ein Item pro Event, unveränderlich. Das ist die Source of Truth, die einzige Wahrheit im System.
Die Streams bilden den Change-Feed dieser Tabelle. Jede neue Zeile erzeugt einen Stream-Record. Kein zweiter Store, nur ein Fenster auf die Änderungen.
Die Projektion ist eine Lambda, die die Stream-Records liest und daraus denormalisierte Read-Models schreibt. Das ist exakt die Projektions-Idee der Read Models, nur eben in AWS-Diensten gegossen.

Wichtig für die spätere Fehlersuche: Der Stream ist kein Archiv. Records werden nach 24 Stunden getrimmt. Wer den kompletten Verlauf braucht, liest ihn aus der Tabelle, nicht aus dem Stream. Der Stream transportiert nur den Delta-Fluss von der Write-Side zur Read-Side.

Im Zusammenhang sieht der Weg eines Commands so aus:

graph LR
  A[Command] --> B[CommandHandler]
  B -->|Conditional Write| C[(DynamoDB<br/>Event-Store)]
  C --> D[DynamoDB Streams]
  D --> E[Lambda<br/>Projektor]
  E --> F[(Read-Model-Tabelle)]
  F --> G[Query]

Links steht die Schreibseite, rechts die Leseseite. Der Stream ist die Naht dazwischen – asynchron, entkoppelt, at-least-once. Genau das ist CQRS: getrennte Modelle für Schreiben und Lesen, verbunden über einen Event-Fluss.

Der Event-Store als Tabelle

Das Datenmodell ist bewusst schlicht. Der Partition-Key ist die aggregateId, der Sort-Key ist die version – zero-padded, damit die numerische Sortierung als String stimmt. Ein Aggregat ist damit eine Partition, und seine Events liegen dort aufsteigend nach Version. Das passt direkt auf die Prinzipien aus dem Single-Table-Design: Der Zugriffspfad steht im Schlüssel.

Streams aktiviere ich beim Anlegen der Tabelle über die StreamSpecification. Der StreamViewType hat genau vier zulässige Werte – KEYS_ONLY, NEW_IMAGE, OLD_IMAGE und NEW_AND_OLD_IMAGES. Für Event Sourcing brauche ich das neue Bild, denn das Event steckt in der frisch geschriebenen Zeile:

Resources:
  EventStore:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: event-store
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: pk
          AttributeType: S
        - AttributeName: sk
          AttributeType: S
      KeySchema:
        - AttributeName: pk
          KeyType: HASH
        - AttributeName: sk
          KeyType: RANGE
      StreamSpecification:
        StreamViewType: NEW_IMAGE

Ein Detail, das mich früher einmal Stunden gekostet hat: Der StreamViewType ist nach dem Anlegen nicht mehr änderbar. Wer ihn wechseln will, muss den Stream deaktivieren und neu anlegen – mit einer neuen Stream-ARN und damit einem neuen Event Source Mapping. Das im ersten Anlauf richtig zu setzen spart später einen unangenehmen Migrationsschritt.

Events anhängen mit Optimistic Concurrency

Der interessante Teil beim Schreiben ist nicht das Schreiben selbst, sondern die Frage: Was, wenn zwei Commands gleichzeitig dasselbe Aggregat auf Version N+1 fortschreiben wollen? Genau einer darf gewinnen. Das löse ich mit Optimistic Concurrency über einen Conditional Write.

Die Idee: Ich schreibe das Event für version = expectedVersion + 1 und knüpfe die Bedingung daran, dass unter diesem Schlüssel noch kein Item existiert. Existiert es schon, hat jemand anderes die Version belegt, und DynamoDB wirft eine ConditionalCheckFailedException. Der Caller liest dann die aktuelle Version neu und versucht den Command erneut.

import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import {
  DynamoDBDocumentClient,
  PutCommand,
} from "@aws-sdk/lib-dynamodb";
import { ConditionalCheckFailedException } from "@aws-sdk/client-dynamodb";

const client = DynamoDBDocumentClient.from(new DynamoDBClient({}));

export class ConcurrencyError extends Error {}

interface DomainEvent {
  type: string;
  payload: Record<string, unknown>;
}

export async function appendEvent(
  aggregateId: string,
  expectedVersion: number,
  event: DomainEvent,
): Promise<void> {
  const nextVersion = expectedVersion + 1;

  try {
    await client.send(
      new PutCommand({
        TableName: "event-store",
        Item: {
          pk: aggregateId,
          sk: String(nextVersion).padStart(12, "0"),
          version: nextVersion,
          type: event.type,
          payload: event.payload,
          createdAt: new Date().toISOString(),
        },
        ConditionExpression: "attribute_not_exists(pk)",
      }),
    );
  } catch (err) {
    if (err instanceof ConditionalCheckFailedException) {
      throw new ConcurrencyError(
        `version ${nextVersion} already exists for ${aggregateId}`,
      );
    }
    throw err;
  }
}

attribute_not_exists(pk) prüft, ob der komplette Item-Key frei ist – DynamoDB wertet die Bedingung gegen genau das Item aus, dessen Schlüssel im PutCommand steht. Damit ist die Kombination aus aggregateId und version der Uniqueness-Anker. Kein zusätzlicher Lock, keine separate Sequenz-Tabelle, keine Transaktion über mehrere Systeme.

Wenn ich mehrere Events atomar schreiben muss – etwa bei einem Command, der zwei Domänen-Ereignisse auslöst – nutze ich TransactWriteItems statt einzelner Puts. Das bündelt bis zu 100 Items in einer atomaren Transaktion, und ich kann die gleiche Bedingung auf jedes Event legen. Für die Regel „ein Event = ein Item“ reicht der einfache Conditional Write.

Die Projektion als Stream-Handler

Jetzt kommt die Leseseite. Sobald ein Event geschrieben ist, erscheint es im Stream, und das Event Source Mapping zieht es in eine Lambda. Deren Aufgabe: die Events auf ein Read-Model falten.

Der Handler bekommt ein DynamoDBStreamEvent mit einem Batch von Records. Jeder Record trägt einen eventName – für unseren append-only Store ist das immer INSERT. Das neue Bild lese ich aus dynamodb.NewImage und unmarshalle es in ein normales Objekt.

import type { DynamoDBStreamEvent, DynamoDBBatchResponse } from "aws-lambda";
import { unmarshall } from "@aws-sdk/util-dynamodb";
import {
  DynamoDBDocumentClient,
  UpdateCommand,
} from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";

const client = DynamoDBDocumentClient.from(new DynamoDBClient({}));

export async function handler(
  event: DynamoDBStreamEvent,
): Promise<DynamoDBBatchResponse> {
  const batchItemFailures: { itemIdentifier: string }[] = [];

  for (const record of event.Records) {
    if (record.eventName !== "INSERT") continue;
    if (!record.dynamodb?.NewImage) continue;

    try {
      const item = unmarshall(record.dynamodb.NewImage as never);
      await project(item);
    } catch (err) {
      batchItemFailures.push({
        itemIdentifier: record.dynamodb!.SequenceNumber!,
      });
    }
  }

  return { batchItemFailures };
}

Zwei Dinge sind hier wichtig. Erstens gebe ich am Ende batchItemFailures zurück. Das aktiviert ReportBatchItemFailures: Statt den ganzen Batch bei einem einzelnen kaputten Record erneut zuzustellen, meldet die Lambda nur die fehlgeschlagenen Records – identifiziert über ihre SequenceNumber. Der Rest gilt als verarbeitet. Ohne dieses Muster blockiert ein einziger Poison-Pill-Record die gesamte Verarbeitung des Shards.

Zweitens die eigentliche Projektion. Hier lauert der teuerste Fehler des ganzen Setups, und deshalb bekommt er einen eigenen Abschnitt.

Idempotenz ist keine Option

Das Event Source Mapping liefert at-least-once. Der Stream selbst enthält jeden Record genau einmal – aber die Verarbeitung durch Lambda kann bei Retries, bei BisectBatchOnFunctionError oder bei einem Timeout denselben Batch mehrfach zustellen. Diese beiden Aussagen werden in Sekundärquellen oft vermischt: Der Stream ist exactly-once und pro Item geordnet; die Lambda-Verarbeitung ist at-least-once. Für den Code, den ich schreibe, zählt die zweite Aussage.

Konkret heißt das: Meine Projektion muss denselben Event zweimal verkraften, ohne das Read-Model zu verfälschen. Bei einer Zählung wäre eine doppelte Verarbeitung sonst eine Doppelzählung. Der Guard dagegen ist ein Conditional Write auf eine monoton steigende Version im Read-Model. Ich schreibe die neue Projektion nur, wenn die verarbeitete Version echt größer ist als die zuletzt gespeicherte:

async function project(item: Record<string, unknown>): Promise<void> {
  const aggregateId = item.pk as string;
  const version = item.version as number;

  await client.send(
    new UpdateCommand({
      TableName: "read-model-orders",
      Key: { pk: aggregateId },
      UpdateExpression: [
        "SET lastProcessedVersion = :v",
        "#status = :status",
        "updatedAt = :now",
      ].join(", "),
      ConditionExpression:
        "attribute_not_exists(pk) OR lastProcessedVersion < :v",
      ExpressionAttributeNames: {
        "#status": "status",
      },
      ExpressionAttributeValues: {
        ":v": version,
        ":status": deriveStatus(item),
        ":now": new Date().toISOString(),
      },
    }),
  );
}

Kommt ein Event doppelt an, schlägt die Bedingung lastProcessedVersion < :v beim zweiten Mal fehl, und das Update passiert nicht. Das ist idempotent – nicht, weil ich Duplikate erkenne und verwerfe, sondern weil ein wiederholter Write schlicht keine Wirkung hat. Wenn die ConditionalCheckFailedException hier auftritt, fange ich sie bewusst ab und behandle sie als Erfolg, nicht als Fehler.

Ein Randfall, der leicht übersehen wird: Ein PutItem oder UpdateItem, das die Daten gar nicht verändert, erzeugt keinen Stream-Record. Wer seine Projektions-Erwartungen an „für jeden Write kommt ein Event“ knüpft, wundert sich sonst über fehlende Records. Im append-only Event-Store tritt das nicht auf, weil jedes Event ein neues Item ist – aber sobald man am Store herummodelliert, sollte man es im Kopf behalten.

Ordering: pro Aggregat, nicht global

Der zweite Fallstrick betrifft die Reihenfolge. AWS garantiert für DynamoDB Streams: Für jedes einzelne Item erscheinen die Stream-Records in derselben Reihenfolge wie die tatsächlichen Änderungen an diesem Item. Das ist genau die Garantie, die Event Sourcing braucht – innerhalb eines Aggregats ist die Event-Folge korrekt geordnet.

Was es nicht gibt, ist globales Ordering über Aggregate hinweg. Die Records verteilen sich auf Shards, und Shards werden parallel verarbeitet. Wer den ParallelizationFactor hochdreht – bis zu 10 nebenläufige Batches pro Shard – verschärft das noch. Zwei Events aus verschiedenen Aggregaten können in beliebiger Reihenfolge in der Projektion ankommen.

In der Praxis ist das selten ein Problem, weil Projektionen fast immer pro Aggregat denken. Wer aber eine global chronologische Sicht braucht – etwa einen Feed über alle Bestellungen sortiert nach Zeit – darf sich nicht auf die Stream-Reihenfolge verlassen, sondern muss über einen expliziten Timestamp oder eine Sequenz im Event sortieren. Das ist eine Modellierungsentscheidung, keine Konfigurationsschraube.

Rebuild und Snapshots

Ein Read-Model ist ableitbar. Das ist einer der schönsten Vorteile des Ansatzes: Wenn ich die Projektionslogik ändere oder ein Read-Model korrupt ist, werfe ich es weg und baue es neu. Der Rebuild ist ein Replay – ich lese alle Events eines Aggregats aufsteigend und falte sie:

import {
  DynamoDBDocumentClient,
  QueryCommand,
} from "@aws-sdk/lib-dynamodb";

export async function rebuildProjection(
  aggregateId: string,
  fromVersion = 0,
): Promise<ReadModelState> {
  const result = await client.send(
    new QueryCommand({
      TableName: "event-store",
      KeyConditionExpression: "pk = :id AND sk > :from",
      ExpressionAttributeValues: {
        ":id": aggregateId,
        ":from": String(fromVersion).padStart(12, "0"),
      },
    }),
  );

  const events = result.Items ?? [];
  return events.reduce(applyEvent, initialState());
}

Bei Aggregaten mit sehr vielen Events wird das teuer – jedes Replay liest den ganzen Stream. Dagegen empfiehlt die AWS Prescriptive Guidance Snapshots: den materialisierten Zustand eines Aggregats zu Version N, periodisch abgelegt. Der Rebuild lädt dann den Snapshot und faltet nur die Delta-Events seit dieser Version. Der fromVersion-Parameter oben ist genau die Naht dafür – ich starte den Query bei snapshot.version statt bei null. Der komplette Rebuild ohne Snapshot bleibt als Reißleine für den Fall, dass sich die Fold-Logik selbst geändert hat.

Hot Partitions und der blockierte Shard

Der dritte Fallstrick ist der Preis der klaren Partitionierung. Die aggregateId als Partition-Key bedeutet: Ein sehr aktives Aggregat konzentriert alle Writes und den gesamten Stream-Verkehr auf eine Partition und einen Shard. Das führt zu Throttling und Latenz – eine klassische Hot Partition.

Dazu kommt ein Reader-Limit: Pro Shard sind maximal zwei Prozesse zulässig, die gleichzeitig lesen. Ein Batch, den die Lambda nicht verarbeiten kann, blockiert außerdem den Fortschritt des Shards, bis er entweder gelingt oder ausläuft. Genau dagegen gibt es die Stellschrauben am Event Source Mapping:

BisectBatchOnFunctionError teilt einen fehlgeschlagenen Batch rekursiv, um den einen giftigen Record zu isolieren.
MaximumRetryAttempts und MaximumRecordAgeInSeconds begrenzen, wie lange ein Record blockieren darf.
Ein On-failure-DestinationConfig schiebt endgültig gescheiterte Batches nach SQS oder SNS, statt sie im Stream festzuhalten.

Zusammen mit ReportBatchItemFailures aus dem Handler ergibt das eine Fehlerbehandlung, die einen einzelnen kaputten Event nicht die ganze Projektion anhalten lässt. Wer das von Anfang an setzt, spart sich den Incident, bei dem eine Read-Model-Tabelle stundenlang stehenbleibt, weil ein einziger Record nicht geparst werden konnte.

Fazit

Event Sourcing auf AWS braucht erstaunlich wenig: eine DynamoDB-Tabelle mit aggregateId als Partition-Key und version als Sort-Key, einen Conditional Write für Optimistic Concurrency, DynamoDB Streams als Change-Feed und eine Lambda, die daraus idempotente Read-Models faltet. Das ist Event Sourcing plus CQRS, serverless zusammengesetzt aus verwalteten Diensten.

Die Konzepte tragen genauso wie auf dem Whiteboard – aber die Betriebsdetails entscheiden, ob das System nachts durchhält. Drei Dinge nehme ich in jedes Projekt mit: Projektionen müssen idempotent sein, weil die Verarbeitung at-least-once ist. Ordering gilt pro Aggregat, nicht global – wer global sortieren will, modelliert einen Timestamp ins Event. Und die Fehlerbehandlung am Event Source Mapping ist kein Optional, sondern der Unterschied zwischen einem selbstheilenden und einem blockierten Stream.

Wer diesen Stack einmal aufgesetzt hat, merkt schnell, wie gut er sich weiterspinnen lässt – etwa in Richtung einer GraphQL-Leseseite mit AppSync und DynamoDB, die direkt auf den projizierten Read-Models sitzt. Der Event-Store bleibt die Wahrheit, die Projektionen bleiben ableitbar, und die API liest nur noch fertig geformte Sichten.