Consumers

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 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; 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. 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 and agents 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

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, from the beginning of the log.

Lanes

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

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

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 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 it.

Run a consumer now with consume

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

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) 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

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

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. 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.

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 once you know why.

When someone presses stop, catch the abort and return normally instead. See Cancellation.

Resolve a dead letter

A dead-lettered event waits for you. List a session's dead letters, then retry or skip each one for the consumer it's dead for:

// 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 }.