Projections from DynamoDB Streams

A projection folds a stream’s events into a read model: an order summary, a balance, a search document. With EventStore, the events usually reach the projection through the table’s change feed: DynamoDB Streams, read by a Lambda event source mapping or an EventBridge Pipe. The change feed does not deliver a stream’s events in version order. EventStore.projection takes the order from the event store instead.

Outside Lambda and Pipes — a long-running service, a container, an edge device — effect-dynamodb/change-feed polls the stream itself and runs it into the same projections.

Delivery order is not event order

• DynamoDB Streams orders records per item only. The DynamoDB Streams documentation says it “guarantees ordering at the level of an individual item … not across an entire partition”. Every event of a stream is its own item (one sort key per version), so nothing orders a stream’s events relative to each other.
• A transaction’s records carry no grouping or order. In the AWS wording, “stream records from the same transaction might appear at different times and could be interleaved with records from other transactions”. An append of several events can arrive split across batches, interleaved with other appends, or out of version order.
• ParallelizationFactor above 1 processes one shard’s records on several concurrent lanes, keeping order only per item. A stream’s events then routinely finish out of order.
• Delivery is at least once. A failed batch is retried, and a record can arrive again after it was processed.

Why waiting for the predecessor stalls

A fold has to apply version n − 1 before n. The usual way to cope with n arriving first is to wait for n − 1 to be delivered: poll the read model, then fail the record so it is retried.

With ParallelizationFactor 1, or whenever n and n − 1 share a lane, n − 1 is queued behind n. So n waits for a record that can’t arrive until n is done, and fails. It is then retried ahead of n − 1 again, until its retries run out and it is dead-lettered. The lane stalls the whole time, and the read model never gets n.

The event store already holds what is missing. An append is one transaction with a version-contiguity check, so when version n has a stream record, versions 1 … n are all committed. A strongly consistent read returns them. A projection never has to wait for a delivery.

Setup

The examples on this page use this stream:

class OrderPlaced extends Schema.TaggedClass<OrderPlaced>()("OrderPlaced", {
  customer: Schema.String,
}) {}
class ItemAdded extends Schema.TaggedClass<ItemAdded>()("ItemAdded", {
  sku: Schema.String,
  price: Schema.Number,
}) {}
class OrderShipped extends Schema.TaggedClass<OrderShipped>()("OrderShipped", {}) {}

const AppSchema = DynamoSchema.make({ name: "shop", version: 1 })
const EventsTable = Table.make({ schema: AppSchema, entities: {} })

const OrderEvents = EventStore.makeStream({
  table: EventsTable,
  streamName: "Order",
  events: [OrderPlaced, ItemAdded, OrderShipped],
  streamId: { composite: ["orderId"] },
})

The read model is an ordinary entity — here on its own table, with a GSI. A numeric field holds the checkpoint: the last stream version applied to the row.

const OrderSummary = Schema.Struct({
  orderId: Schema.String,
  customer: Schema.String,
  status: Schema.String,
  items: Schema.Number,
  total: Schema.Number,
  version: Schema.Number, // the checkpoint: the last applied stream version
})

const OrderSummaries = Entity.make({
  model: OrderSummary,
  entityType: "OrderSummary",
  primaryKey: {
    pk: { field: "pk", composite: ["orderId"] },
    sk: { field: "sk", composite: [] },
  },
  indexes: {
    byStatus: {
      name: "gsi1",
      pk: { field: "gsi1pk", composite: ["status"] },
      sk: { field: "gsi1sk", composite: ["orderId"] },
    },
  },
})

Defining a projection

A projection binds one source stream to a read model:

const OrderSummaryProjection = EventStore.projection({
  stream: OrderEvents,
  readModel: OrderSummaries,
  key: (streamId) => ({ orderId: streamId.orderId }),
  initial: (streamId) => ({
    orderId: streamId.orderId,
    customer: "",
    status: "new",
    items: 0,
    total: 0,
    version: 0,
  }),
  evolve: (state, event) => {
    const data = event.data
    switch (data._tag) {
      case "OrderPlaced":
        return Effect.succeed({ ...state, customer: data.customer, status: "placed" })
      case "ItemAdded":
        return Effect.succeed({ ...state, items: state.items + 1, total: state.total + data.price })
      case "OrderShipped":
        return Effect.succeed({ ...state, status: "shipped" })
    }
  },
  checkpoint: "version",
})

• key(streamId, event) is the row an event folds into. A key built from the stream id alone gives one row per stream instance; a key that also reads the event lets a stream feed several rows (see Rows keyed by event fields). streamId is structured: a decoded event carries it as the composites joined with # (StreamEvent.streamId), and the projection splits it back with stream.parseStreamId — on a multi-field id, a value containing # is ambiguous and fails with ValidationError.
• An event can address no row: return an Option from key, and Option.none() acknowledges the event — no row is read or written, no checkpoint moves, and applyBatch lists its position in ignored. apply then returns an Option too, and catchUp targets the latest event that does address a row.
• initial(streamId, event) is the state a row starts from, before event — the first event folded into a row that does not exist yet.
• evolve returns an Effect. Returning the state unchanged still writes the row and advances the checkpoint. Return EventStore.skipEvent instead to leave the row alone. A skipped event is reported in skipped; a row whose events are all skipped isn’t written; and the checkpoint stays below the skipped event, which is decided again later (see Skipping an event). A failure fails that row only — unless terminal accepts it (below). evolve runs again on every lost-race retry, so keep it free of side effects.
• checkpoint names a numeric field, written by the library after the fold. { field: "version", offset: -1 } stores version + offset — for a read model that numbered versions differently (here, from 0) — with reads and conditions going through the same mapping, so stored values and their readers stay as they were. The field may be optional (Schema.optional) for rows written before the projection: an absent checkpoint applies every event, and the first write is conditioned on attribute_not_exists.
• The projection owns the row. It reads it strongly consistent — Hidden fields included — and writes it whole with put (or create for a new row), keeping its createdAt. toItem(state, { streamId, key }) maps the folded state to the row as written, to recompute attributes kept for other readers.
• Refused read models throw [EDD-9081]: versioned (the checkpoint is the version), softDelete (projections never delete rows) and timeSeries.

Handling a batch

processBatch takes the batch’s raw records, decodes the INSERTs of this stream’s events, applies them, and builds the partial batch response:

type StreamBatch = { readonly Records: ReadonlyArray<EventStore.StreamRecord> }

const makeHandler = Effect.gen(function* () {
  const summaries = yield* EventStore.bindProjection(OrderSummaryProjection)
  return (batch: StreamBatch) =>
    Effect.gen(function* () {
      const result = yield* summaries.processBatch(batch.Records, { source: "lambda" })
      for (const outcome of result.outcomes) {
        if (outcome._tag === "Failed") continue
        // Side effects for the events this call applied, in version order
        for (const event of outcome.applied) yield* publish(outcome.streamId, event)
      }
      return result.response
    })
})

EventStore.bindProjection resolves what the projection needs — the client, both tables’ TableConfig, evolve’s requirements and a Compression registry for packed event data — so every operation has R = never.

Underneath is applyBatch(events), which you can call with decoded events. Per row (concurrently across rows, concurrency 10 by default):

1. Sort and deduplicate the row’s events by version — an out-of-order batch, or one carrying a redelivery, needs no read to put right.
2. Read the row (strongly consistent) and its checkpoint c. Events at or below c are reported as alreadyApplied; with nothing left, the row is done.
3. Fill gaps from the store. If the batch holds every version from c + 1 to the highest delivered one — the row’s own events, or the stream’s events for other rows or for no row — nothing is read. Otherwise one strongly consistent readFrom(c, { toVersion }) — bounded at that highest version — supplies the missing ones.
4. Fold and write once, conditioned on the checkpoint still being c: k events of a row in the batch cost one write.

A redelivered or late event is skipped by the checkpoint; a missing earlier one is read from the store. Nothing waits. Outcomes are Applied or Failed per row; applyBatch itself never fails, so one row’s error can’t hide the others’ applied. apply(event) is applyBatch([event]) with the row’s failure in the error channel. The outcome positions of processBatch index the records you passed.

Side effects

Each outcome’s applied lists the events that call applied, in version order — those read from the store to fill a gap included. Every event appears in exactly one call’s applied: the one whose conditional write won.

• applied is at most once. A crash after the write and before publishing loses the side effect: the redelivery is skipped by the checkpoint.
• alreadyApplied opts into at least once. It lists the batch’s events the row already had; republish them, with consumers tolerating duplicates (by stream id and version, say). With gap-filling, an event applied early (by a later event’s fill) is delivered again in its own right and lands in alreadyApplied, so this republishes on every reordering, not only after a failure.
• Order holds within one call’s applied, not across concurrent calls.

Rows keyed by event fields

A key that reads the event lets one stream feed several rows — here one row per sensor and zone. Each row is still fed by that one stream instance, and keeps its own checkpoint: every stream version up to it has been considered for the row. When an event addresses another row, the earlier row keeps what it had, and the new row folds only its own events; a gap read keeps only the row’s events.

class SensorPlaced extends Schema.TaggedClass<SensorPlaced>()("SensorPlaced", {
  zone: Schema.String,
}) {}
class SensorReading extends Schema.TaggedClass<SensorReading>()("SensorReading", {
  zone: Schema.String,
  celsius: Schema.Number,
}) {}

const SensorEvents = EventStore.makeStream({
  table: EventsTable,
  streamName: "Sensor",
  events: [SensorPlaced, SensorReading],
  streamId: { composite: ["sensorId"] },
})

// One row per sensor and zone: the zone comes from each event
const ZoneReadings = Entity.make({
  model: Schema.Struct({
    sensorId: Schema.String,
    zone: Schema.String,
    readings: Schema.Number,
    maxCelsius: Schema.Number,
    seq: Schema.Number,
    alertKey: DynamoModel.Hidden(Schema.String), // kept for another reader
  }),
  entityType: "ZoneReading",
  primaryKey: {
    pk: { field: "pk", composite: ["sensorId"] },
    sk: { field: "sk", composite: ["zone"] },
  },
})

class ZoneNotPlaced extends Data.TaggedError("ZoneNotPlaced")<{ readonly version: number }> {}

const ZoneReadingProjection = EventStore.projection({
  stream: SensorEvents,
  readModel: ZoneReadings,
  key: (streamId, event) => ({ sensorId: streamId.sensorId, zone: event.data.zone }),
  initial: (streamId, event) => ({
    sensorId: streamId.sensorId,
    zone: event.data.zone,
    readings: -1, // not placed yet
    maxCelsius: Number.NEGATIVE_INFINITY,
    seq: 0,
    alertKey: "",
  }),
  evolve: (state, event) =>
    Effect.gen(function* () {
      const data = event.data
      if (data._tag === "SensorPlaced") return { ...state, readings: 0 }
      if (state.readings < 0) return yield* new ZoneNotPlaced({ version: event.version })
      return {
        ...state,
        readings: state.readings + 1,
        maxCelsius: Math.max(state.maxCelsius, data.celsius),
      }
    }),
  checkpoint: "seq",
  // A reading of a zone never placed is acknowledged and skipped, every time
  terminal: (error) => error._tag === "ZoneNotPlaced",
  // Recompute the attribute other readers query by, on every write
  toItem: (state) => ({ ...state, alertKey: state.maxCelsius > 30 ? "hot" : "normal" }),
})

• terminal marks the evolve failures that reject their event rather than fail the row. A rejected event is skipped and reported in the outcome’s rejected; the fold continues, and rejected events alone never create a row (its state is Option.none()). On an existing row, the checkpoint advances past it. The decision must be deterministic: a rejected event of a row that doesn’t exist yet is read again to fill a later gap — and rejected again.
• A row that is never created has no checkpoint, so each later delivery of its events gap-reads from version 1 and reports the earlier rejections again. Dedupe rejected by version when you count or log them; the read grows with the stream.
• Without terminal, a failure fails the row on every delivery: the checkpoint stays behind the failing event, so each later event of the row gap-fills it and fails the same way.
• toItem recomputes a Hidden attribute other readers use, on every write.

Partial batch responses

processBatch reports the earliest record of every failed row, plus every record whose image failed to decode. Enable ReportBatchItemFailures on the event source mapping: Lambda checkpoints at the lowest reported record and retries everything after it, which the checkpoints then skip.

The record identifier depends on who reads the response:

source	Identifier	Consumer
"lambda"	dynamodb.SequenceNumber	Lambda event source mapping (docs)
"pipe"	eventID	EventBridge Pipe target (docs)

• A Pipe honours partial batch responses from its target, not from an enrichment.
• A record shape of your own — a Pipe input template’s, say — is read with image: (record) => … and identifier: (record) => … instead of source.
• A failed record without an identifier fails the call with StreamRecordIdentifierMissing (an empty identifier would fail the whole batch silently): fail the invocation, and the source redelivers the batch.
• A record that fails to decode fails the same way every time, so reporting it holds its shard until the source gives up. onDecodeFailure: "skip" acknowledges it instead — it is still in decodeFailures, to log and dead-letter yourself. That acknowledges the record, not the stored event: the next event of the same row gap-fills through it, fails to decode it, and fails the row. Skipping helps when the bad event is its stream’s last, or once the stored item is repaired or removed.
• To fail more records (a side effect that failed, say), add them to result.failedRecords and call EventStore.batchResponse(records, { source }).

One feed, several projections

A table’s feed usually carries several streams. EventStore.processBatch decodes each record once, routes it to every projection of its stream — records of other streams and items are skipped — and answers with one response across them all:

const makeFeedHandler = Effect.gen(function* () {
  const summaries = yield* EventStore.bindProjection(OrderSummaryProjection)
  const zones = yield* EventStore.bindProjection(ZoneReadingProjection)
  return (batch: StreamBatch) =>
    EventStore.processBatch(batch.Records, {
      projections: { summaries, zones },
      source: "lambda",
    }).pipe(Effect.map((result) => result.response))
})

Each projection’s outcomes come back under its name, with positions indexing the records. To act on a stream’s events without decoding twice (side effects only), name it in streams: { Name: stream }. Its events come back in unrouted, each with its record’s position, whether or not a projection also reads the stream. Two streams storing events under one marker make the call die with [EDD-9082].

A process that polls the stream itself runs the same call per batch with runChangeFeed, which acknowledges each batch per its response.

Decoding several streams and filtering the feed

EventStore.decoder builds one decoder for several streams: it dispatches on the image’s __edd_e__, and returns Option.none() for snapshots, idempotency sentinels and other items. EventStore.eventFilterPattern builds the filter that keeps only these streams’ event INSERTs, so nothing else invokes the function:

const decode = EventStore.decoder({ Order: OrderEvents })

const describeImage = (image: Record<string, AttributeValue>) =>
  decode(image).pipe(
    Effect.map(
      Option.match({
        onNone: () => "not an event",
        onSome: ({ stream, event }) => `${stream} v${event.version}`,
      }),
    ),
  )

// FilterCriteria for the event source mapping (or the Pipe)
const filterCriteria = {
  Filters: [{ Pattern: JSON.stringify(EventStore.eventFilterPattern([OrderEvents])) }],
}

The decoded value’s stream is the name you gave the stream in the record, so a switch on it narrows event. Two streams that store events under the same marker throw [EDD-9082].

Catch-up and rebuild

const caughtUp = yield* OrderSummaryProjection.catchUp({ orderId: "o-1" })
const rebuilt = yield* OrderSummaryProjection.rebuild({ orderId: "o-1" })

• catchUp(streamId) brings the row of the stream’s latest event up to the head — for a stream whose last event was dead-lettered (an earlier missing event is filled in by the next delivery anyway). With one row per stream instance, that is the only row.
• rebuild(streamId) re-derives every row of the stream instance from version 1, ignoring their state — for backfills, a new read model over existing streams, or a changed evolve — and returns no events, so it never replays side effects. Without a revision, pause live delivery for the instance while it runs: the write is then conditioned on the checkpoint. For a row already at the head, the checkpoint doesn’t change, so an applyBatch that read the row before the rebuild can still overwrite it. A revision changes on every write, which closes that gap.

Neither reads the stream’s snapshot: that is the decider’s state, not the read model’s.

Tuning

With gap-filling, ParallelizationFactor and BatchSize decide throughput and latency, not correctness:

• A larger BatchSize puts more of a row’s events in one batch, ordered in memory and written once.
• A higher ParallelizationFactor runs more lanes, and calls that touch one row race on the conditional write. Each round has one winner; a loser re-reads, and has nothing left to do only when the winner reached its events. The default retry waits a jittered, growing delay up to 12 times (about 1.3 s in all); raise it if many lanes write the same rows. Exhausted, the row fails with ConcurrentModification and its records go back to the source — safe, as the redelivery is skipped or gap-filled.

Beyond one source

Multi-source Projections, Tombstones and the Outbox covers:

• rows fed by several streams;
• revision;
• removing rows;
• an ordered, at-least-once outbox.

Limits

• A row of projection is fed by one stream instance. A row that several streams contribute to is a multi-source projection.
• Rows are removed with tombstones (removal), never deleted. Deleting a row would lose its checkpoint, and a redelivery would recreate it.
• projection takes an unbound stream and entity; bind it with EventStore.bindProjection.