Event Sourcing auf AWS: DynamoDB Streams und Lambda

Ein Event-Store ist keine Tabelle, deren Zeilen man aktualisiert. Es wird nur angehängt: Jede Zustandsänderung ist ein neues, unveränderliches Item, und der Zustand eines Aggregats entsteht erst, wenn man dessen Events in Versionsreihenfolge faltet. In der Konzept-Betrachtung zum Event Sourcing habe ich das als Logbuch beschrieben – die Ereignisfolge trägt die Wahrheit, der Zustand ist nur ihr Ergebnis.

Auf dem Whiteboard ist das schnell erklärt. Tragfähig wird es erst mit den Antworten auf die nächsten Fragen: Wo läuft der Store? Wer garantiert die Reihenfolge? Was passiert, wenn ein Consumer denselben Event zweimal sieht? Auf AWS lässt sich das aus verwalteten Diensten zusammensetzen, und die scharfen Kanten liegen dann bei Nebenläufigkeit beim Anhängen, bei der Reihenfolge im Change-Feed und bei doppelter Zustellung an die Projektion. Genau darum geht es hier – samt dem Code, den 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 verbindet beide Seiten asynchron. DynamoDB Streams hält jeden Änderungsdatensatz einmal vor; die Verarbeitung durch Lambda ist dagegen at-least-once. Genau deshalb muss die Projektion Wiederholungen vertragen.

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,
  GetCommand,
  UpdateCommand,
} from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";

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

class ProjectionGapError extends Error {}

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. Wenn ReportBatchItemFailures am Event Source Mapping aktiviert ist, verwendet Lambda die kleinste gemeldete SequenceNumber als Checkpoint und wiederholt ab dort. Das reduziert Wiederholungen, schließt sie für bereits erfolgreiche Folgerecords aber nicht aus. Ohne partielle Rückmeldung gilt dagegen der ganze Batch als fehlgeschlagen.

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 lückenlos steigende Version im Read-Model. Ich schreibe die neue Projektion nur, wenn ihre Version genau auf die zuletzt gespeicherte folgt:

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

  const current = await client.send(
    new GetCommand({
      TableName: "read-model-orders",
      Key: { pk: aggregateId },
      ConsistentRead: true,
    }),
  );

  const lastProcessedVersion =
    (current.Item?.lastProcessedVersion as number | undefined) ?? 0;

  // A repeated delivery has no effect.
  if (version <= lastProcessedVersion) return;

  // A later event must wait until the missing versions were applied.
  if (version !== lastProcessedVersion + 1) {
    throw new ProjectionGapError(
      `expected version ${lastProcessedVersion + 1}, received ${version}`,
    );
  }

  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 = :previous",
      ExpressionAttributeNames: {
        "#status": "status",
      },
      ExpressionAttributeValues: {
        ":v": version,
        ":previous": lastProcessedVersion,
        ":status": deriveStatus(current.Item?.status, item),
        ":now": new Date().toISOString(),
      },
    }),
  );
}

Kommt ein Event doppelt an, beendet die Funktion die Arbeit vor dem Update. Kommt dagegen Version 5, während im Read Model erst Version 3 steht, wird die Lücke als Fehler gemeldet. Lambda stellt diesen Record erneut zu; Version 4 kann vorher verarbeitet werden. Die Bedingung im UpdateCommand schützt zusätzlich gegen zwei gleichzeitig arbeitende Projektoren. Schlägt sie nach dem Lesen fehl, wird ebenfalls wiederholt – sie darf nicht pauschal als erfolgreicher Duplikatfall verschwinden.

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 muss die Projektion selbst prüfen

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. In diesem Datenmodell ist aber jede Event-Version ein eigenes Item – gleicher Partition-Key, anderer Sort-Key. Aus der Garantie pro Item folgt daher keine garantierte Reihenfolge über die Events eines Aggregats.

Im Normalbetrieb kommen aufeinanderfolgende Writes häufig in der erwarteten Reihenfolge an. Darauf darf sich der Projektor trotzdem nicht verlassen, erst recht nicht bei Retries, partiellen Batch-Fehlern oder einem erhöhten ParallelizationFactor. Deshalb prüft der Code oben auf eine lückenlose Version. Eine Lücke wird wiederholt oder gezielt aus dem Event-Store aufgefüllt, statt ein späteres Event vorzuziehen und das fehlende anschließend als veraltet zu verwerfen.

Globales Ordering über Aggregate hinweg gibt es ebenfalls nicht. Die Records verteilen sich auf Shards, und Shards werden parallel verarbeitet. Wer eine global chronologische Sicht braucht – etwa einen Feed über alle Bestellungen sortiert nach Zeit –, muss eine dafür geeignete Ordnung modellieren. Ein Zeitstempel kann zur Darstellung dienen, ist aber allein noch keine konfliktfreie globale Sequenz.

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. Die Reihenfolge der getrennten Event-Items muss über ihre Version geprüft werden. Und die Fehlerbehandlung am Event Source Mapping entscheidet darüber, ob sich der Stream selbst wieder freiräumt oder stehen bleibt.

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.

Weiterführende Quellen

• Foliensatz zu diesem Thema: https://supabase.mikebild.dev/storage/v1/object/public/slides/introduction-cqrs-event-sourcing.html
• DynamoDB Streams – Retention, StreamViewType, Ordering, Shards
• DynamoDB Streams und Lambda-Trigger
• Using AWS Lambda with DynamoDB – ESM, Batch, ParallelizationFactor, Fehlerbehandlung
• Condition Expressions
• Optimistic Locking mit Versionierung
• AWS Prescriptive Guidance – Event Sourcing Pattern
• AWS Database Blog – Build a CQRS event store with Amazon DynamoDB

Wie fandest du diesen Beitrag?

Kommentare