---
title: Consumers
description: Named readers of every session's log. Each keeps its own place, runs in its own lanes, and retries its own failures.
---

```ts server/consumers.ts
import { createServer } from 'experimental-a2/server'
import { orders } from '@/contracts'

export const ordersServer = createServer({
  contract: orders,
  consumers: [
    {
      name: 'notify-shop',
      types: ['created'],
      run: async ({ event }) => {
        // your side effect, e.g. email the shop, idempotent via event.id:
        // await sendEmailToShop(event.payload.shopId, { idempotencyKey: event.id })
        return { type: 'shop.notified', payload: {} }
      },
    },
  ],
})
```

A consumer runs when an event it takes lands: a plain async function that can
call a database, hit an API, or stream a model response. A2 keeps a durable
cursor for it in every session, claims the event while `run` works, and
retries it after a crash.

Each entry of `createServer({ consumers })` is a consumer: its `name`, the
`types` it takes, and `run`, which `event` narrows to those types.
`createServer` binds the contract to a [store](/stores/overview) and registers
them. `consumer({ contract, ... }).run(fn)` from `experimental-a2/server`
defines one on its own, to keep in its own file, to take from a library, or to
run with [`consume`](#run-a-consumer-now-with-consume); list it the same way.
Two consumers with the same name throw at construction. `experimental-a2/server` is the only entry point
that touches a store, and it fails loudly in a browser bundle.

## Its name is its identity

A2 keys every cursor on the consumer's name. Rename a consumer and A2 treats it
as new: it starts without the old one's place, failures, or dead letters.
Remove one from `consumers` and its cursors wait: nothing claims them, and they
resume where they stopped when the name comes back. During a rolling deploy,
each deployment runs only the consumers it has. A consumer with a string lane
reads the log, so it also picks up events a deployment without it appended, the
next time a deployment that has it appends an event it takes or
[consumes it](#run-a-consumer-now-with-consume). A consumer with a lane
function, or with no lane, gets work only for events appended by a deployment
that has it: their lanes resolve at append.

A consumer's lane is part of its identity too. To change it, or to switch
between a string and a function, give the consumer a new name.

Names are up to 128 letters, digits, and `. _ : -`, starting with a letter or
digit. Names starting with `a2.` belong to A2: [actors](/actors/introduction)
and [agents](/agents/introduction) run on A2's own consumers, and their dead
letters name them, as `a2.actor` or `a2.ai.tools`.

## Which events it takes

`types` lists the event types a consumer runs on. Leave it out to take every
type. An event no consumer takes is still logged, folded into state, and
streamed; it starts no work, with no claim or scheduler message.

Changing `types` applies to events the consumer hasn't looked at yet. A cursor
with work looks at events as it reaches them; an idle one looks at each batch
as it's appended. To run a consumer over history with new types, give it a new
name and `backfill`.

## Servers without consumers

```ts server/order-history.ts
import { createServer } from 'experimental-a2/server'
import { postgres } from 'experimental-a2/store-postgres'
import { orders } from '@/contracts'

export const orderHistory = createServer({
  contract: orders,
  store: postgres({ connectionString: process.env.DATABASE_URL }),
})
```

Consumers are optional. Without any, a server is a durable event log:
`append`, `history`, reducers, and live streams. Events settle in their append
transaction, with no drain, claim, or queue message. A consumer you add later
starts with the events appended after it exists, or with
[`backfill`](#start-from-the-beginning-with-backfill), from the beginning of
the log.

## Lanes

```ts server/lanes.ts
import { consumer, createServer } from 'experimental-a2/server'
import { orders } from '@/contracts'

// one driver at a time, drivers in parallel
const dispatch = consumer({
  contract: orders,
  name: 'dispatch',
  types: ['driver.notified'],
  lane: ({ event }) => `driver:${event.payload.driverId}`,
}).run(async ({ event }) => {
  // your work for this driver, using event.payload.driverId
})

export const lanesServer = createServer({
  contract: orders,
  consumers: [
    // every event it takes, one at a time, in log order
    {
      name: 'order-log',
      lane: 'orders',
      run: async ({ event }) => {
        // your write, keyed by event.index so a retry overwrites it:
        // await db.upsert({ id: event.index, type: event.type })
      },
    },
    dispatch,
  ],
})
```

Without `lane`, a consumer runs every event concurrently, each on its own
cursor: log order stays stable, run order doesn't, and a failure retries that
event without holding up the rest. Use a lane when events share a resource and
must not overlap. A string lane runs everything the consumer takes in log
order, one at a time. A function returns each event's lane, or `null` to run
that event concurrently.

A2 resolves each consumer's lane when the event is appended and stores it, so
a deployment can't reinterpret pending work. Lanes belong to one session and
one consumer: they aren't locks across sessions, and two consumers using the
same lane name never wait on each other. Within a lane, a failing event holds
up the events behind it until it succeeds or you resolve it.

## One event, many consumers

```ts server/fanout.ts
import { createServer } from 'experimental-a2/server'
import { orders } from '@/contracts'

export const fanoutServer = createServer({
  contract: orders,
  consumers: [
    {
      name: 'email-shop',
      types: ['created'],
      run: async ({ event }) => {
        // await sendEmailToShop(event.payload.shopId, { idempotencyKey: event.id })
      },
    },
    {
      name: 'reserve-stock',
      types: ['created'],
      run: async ({ event }) => {
        // await reserve(event.payload.items, { idempotencyKey: event.id })
      },
    },
  ],
})
```

Every consumer that takes an event gets its own cursor for it. They run
concurrently, retry separately, and dead-letter separately: an email that keeps
failing never stops the stock reservation.

## Start from the beginning with `backfill`

```ts server/backfill.ts
import { consumer, createServer } from 'experimental-a2/server'
import { orders } from '@/contracts'

const ordersTable = consumer({
  contract: orders,
  name: 'orders-table-v1',
  lane: 'orders-table',
  backfill: true,
}).run(async ({ event }) => {
  // your table, written idempotently by event index:
  // await db.insert(ordersTable).values({ id: event.index, type: event.type }).onConflictDoNothing()
})

export const tableServer = createServer({
  contract: orders,
  consumers: [ordersTable],
})
```

A new consumer starts at the first event appended after it exists. With
`backfill: true`, it starts at the beginning of the log instead, so it sees
every event already there. Use it to fill a table or index from history, then
keep it current, as [SQL projections](/server/sql-projections) does. Change what
it writes, give it a new name, and it fills the new shape from the start.

`backfill` needs a string `lane`: a backfill reads the log in order. It can't
use `abortOn`: cancellation is for work in progress, and checking it while
replaying history would read the rest of the log for every event. A
consumer starts in a session the next time an event it takes is appended
there, or when you [consume](#run-a-consumer-now-with-consume) it.

## Run a consumer now with `consume`

```ts server/read.ts
import { consumer, createServer } from 'experimental-a2/server'
import { orders } from '@/contracts'

const ordersTable = consumer({
  contract: orders,
  name: 'orders-table-v1',
  lane: 'orders-table',
  backfill: true,
}).run(async ({ event }) => {
  // await db.insert(ordersTable).values({ id: event.index, type: event.type }).onConflictDoNothing()
})

export const readServer = createServer({
  contract: orders,
  consumers: [ordersTable],
})

export async function readOrder(orderId: string): Promise<void> {
  // for an order the caller just changed, before reading the table:
  await readServer.session(orderId).consume(ordersTable)
  // every event of the order is in your table now:
  // return db.select().from(ordersTable).where(eq(ordersTable.order, orderId))
}
```

`session.consume(consumer)` runs a consumer with a string `lane` on that
session, in this invocation, until it has handled every event it takes up to
the session's head. Pass `{ through: index }` to stop sooner, as soon as an
event you know about is in. It returns `{ index }`: the consumer has handled
every event it takes through that index.

- **Read your writes.** A table a consumer keeps lags its log a little.
  Consume it before a read that has to include a write you just made.
- **Start a backfill now.** A new consumer otherwise starts with the next
  append. `consume(ordersTable, { through: 0 })` creates its cursor and runs
  nothing here; the server carries on with the backfill in the background.
- **Wait in another consumer.** `ctx.session.consume(ordersTable, { through:
  ctx.event.index })` inside one consumer waits until another's table has the
  triggering event, and then reads it. A consumer can't consume itself. Nor
  can it wait on a consumer that is waiting on it: that `consume` throws,
  unless what it asks for is in already.

Consuming a session with nothing in it writes nothing, so a read can consume
whatever session it's asked about.

The server's own consumer keeps one cursor, so `consume` shares it with the
background: when another invocation holds the next event, `consume` waits for
it. It runs nothing past `through`, which carries on in the background. A
consumer the server doesn't run is pulled only by `consume`, and keeps its
place between calls, like a reader's "since I last looked".

Failures count against the same ten-attempt budget as everywhere else, and
`consume` waits out the backoff between them. Once the next event
dead-letters for that consumer, `consume` throws `CONSUMER_DEAD_LETTERED`.
Pass `{ signal }` to stop waiting sooner. An event running here when it fires
finishes on its own.

## The context

| Property      | What it is |
| ------------- | ---------- |
| `ctx.event`   | The triggering event: `{ id, type, payload, index, sessionId, createdAt }`. With `types`, its type narrows to those. |
| `ctx.attempt` | The durable 1-based claim of this event for this consumer. |
| `ctx.session` | This session's handle: `id`, `append`, `schedule`, `clear`, `history`, `state`, `stream`, `consume`, and `setPresence` (presence contracts only). |
| `ctx.signal`  | An `AbortSignal`. Fires on matching `abortOn` events and when the attempt loses its claim (`CLAIM_EXPIRED`, `SUPERSEDED_ATTEMPT`). See [Cancellation](/server/cancellation). |

`ctx.attempt` counts claims of one event: the next event in a lane starts at
1 again. It can skip a number when a process dies between the claim and
`run`.

[`ctx.session.state(reducer)`](/contract/state) folds through
`ctx.event.index`, so later appends don't change what this event sees. Pass
`{ through: 'latest' }` to read the current committed frontier instead.

## Return what happens after success

```ts server/return-many.ts
import { createServer } from 'experimental-a2/server'
import { orders } from '@/contracts'

export const returnServer = createServer({
  contract: orders,
  consumers: [
    {
      name: 'start-shop',
      types: ['shop.started'],
      run: async () => {
        return orders.batch(
          { type: 'shop.notified', payload: {} },
          { type: 'driver.notified', payload: { driverId: 'driver-7' } },
        )
      },
    },
  ],
})
```

Return one event or an array when it should exist only if `run` succeeds. A2
moves the consumer's cursor past the event and appends the batch in one atomic
operation, at consecutive positions. If `run` throws or the process dies
first, none of it exists. Returned events get deterministic ids from the
trigger, the consumer's name, and their position, so a retry after a lost
acknowledgement doesn't duplicate them.

`contract.batch(...)` is a type helper that keeps each event's literal type;
A2 still validates at commit.

## Append what should happen now

```ts server/append-now.ts
import { createServer } from 'experimental-a2/server'
import { orders } from '@/contracts'

export const appendServer = createServer({
  contract: orders,
  consumers: [
    {
      name: 'accept-order',
      types: ['created'],
      run: async ({ session }) => {
        await session.append('notify-shop', { type: 'shop.notified', payload: {} })

        // A consumer of shop.notified may already be running here.
        // await continueCurrentWork()
      },
    },
  ],
})
```

`ctx.session.append(name, ...events)` commits immediately, so the appended
event can start while `run` is still working. Use it for streaming progress,
independent fan-out, or any fact that stays true even if `run` later fails.
Each call is atomic.

The first argument names the append within the triggering event. A2 derives
the ids of events without an `id` from the trigger, the consumer's name, the
append's name, and each event's position, so a retry gets the existing events
back, whatever order parallel branches finish in. Keep the name stable;
reusing it with a different batch conflicts. An explicit `id` wins, so
different triggers can converge on one fact. Top-level
`server.session(id).append(...events)` takes no name.

## Schedule what happens later

`ctx.session.schedule(name, timing, ...events)` appends events later, through
the [scheduler](/schedulers/overview). In a consumer, a delay counts from the
triggering event's `createdAt`, and the timer's identity includes the trigger
and the consumer's name, so a retry arms the same timer. See
[Timers](/server/timers).

## Remember facts in the log

Reducer snapshots are a cache: A2 may discard one and fold the log again. When
`run` learns something a later run needs, such as "this thread was already
triaged", append it as an event and read it back with a reducer. Then a
rebuilt snapshot still knows it, and nothing runs twice to find out.

## Consumers can run twice

Consumers are at-least-once. A crash after your side effect but before A2
records completion runs the event again.

- **External side effects take an idempotency key.** `event.id` is stable
  across attempts and unique within its session. Ids your callers pick, and
  the ids A2 derives from them, can repeat in another session: when yours can,
  add the session to the key, and the contract's name if several contracts
  share it, as in `` `orders:${event.sessionId}:${event.id}` ``.
- **Named appends are deduplicated.** A retry gets the existing events back.
- **Returned events commit with completion.** They all exist with the
  cursor's advance, or none do.

## When it throws

Throwing means "retry me." A2 records the failure on that consumer's cursor and
retries with backoff. After ten failed attempts the event dead-letters for that
consumer: its lane waits, while other consumers and lanes continue.
[Retry or skip it](#resolve-a-dead-letter) once you know why.

When someone presses stop, catch the abort and return normally instead. See
[Cancellation](/server/cancellation).

## Resolve a dead letter

A [dead-lettered](/server/guarantees#when-a-consumer-keeps-failing) event waits
for you. List a session's dead letters, then retry or skip each one for the
consumer it's dead for:

```ts
// anywhere on the server, behind your own operator checks:
import { ordersServer } from '@/server'

const order = ordersServer.session(orderId)
const [dead] = await order.deadLetters()
if (dead) {
  // dead.lastError says why (null if every attempt timed out or crashed);
  // deploy the fix first, then:
  await order.redrive(
    { consumer: dead.consumer, index: dead.index },
    { action: 'retry' },
  )
}
```

- **Retry.** `{ action: 'retry' }` runs the consumer again with a fresh
  ten-failure budget. Once it completes, its lane continues.
- **Skip.** `{ action: 'skip' }` moves the consumer past the event without
  running it, accepting that its effects never happen for that consumer. The
  next event in its lane runs.

Each redrive is one atomic store operation, and it wakes the session like an
append, so the lane resumes without another write. An event that isn't
dead-lettered anymore is left alone: `redrive` reports `{ redriven: false }`.