# Sinks (/docs/extending/sinks)

> Event and audit sinks receive every committed row at least once, each with its own durable cursor. How dispatch works, the cursor pitfalls, poison events, and usage.



Every change in Better Compute writes an event to the outbox in the same transaction as the
change. An **event sink** is how you get all of those events out of the database and into something else: a
warehouse, a queue, a platform-wide audit feed. The control plane hands each sink every committed event, in commit
order, at least once, and remembers per sink how far it got.

## The contract [#the-contract]

```ts title="@better-compute/core"
interface EventSink {
  readonly extension: 'sink';
  readonly kind: 'events';
  readonly id: string;
  /** At-least-once; events carry ids for deduplication. */
  deliver(events: readonly ComputeEvent[]): Promise<void>;
}

interface ComputeEvent<D = Json> {
  id: string;       // evt_…, stable across redeliveries
  type: string;     // reservation.placed, node.connected, …
  tenantId: string;
  subject: string;  // reservation/rsv_…
  time: string;     // RFC 3339, from compute_now() at insert
  sequence: string; // <txid>:<seq>, the outbox position
  data: D;
}
```

`deliver` resolves when the batch is safely stored on your side, and throws when it is not. That is the whole
protocol.

```ts title="compute.ts"
import { betterCompute, defineSink } from 'better-compute';

const warehouse = defineSink({
  kind: 'events',
  id: 'warehouse',
  async deliver(events) {
    // One idempotent write per batch; event.id is the deduplication key.
    await warehouseClient.upsert('compute_events', events.map((event) => ({ key: event.id, row: event })));
  },
});

export const compute = betterCompute({
  // …database, secret, baseURL, iam
  sinks: { events: [warehouse] },
});
```

The id must be a non-empty string, unique among the event sinks (two sinks with one id would share a cursor, so the
options refuse it with `INVALID_CONFIG`). Pick it once and keep it: the cursor is stored under the id.

A module or plugin can ship a sink instead: an event sink it lists in its `extensions` joins `sinks.events` and is
delivered to the same way (the same uniqueness check applies across both lists).

## How dispatch works [#how-dispatch-works]

OutboxDispatch runs in the `sinks` worker role with one lease and one loop per sink:
`singleton:outbox-dispatch:<sink id>` (a `h-` plus sha256 prefix for an id that is not lease-safe, that is, not 1 to
100 of letters, digits and `. _ @ -`). Exactly one process delivers to each sink at a time, different `sinks` workers
can deliver different sinks, and a slow sink never delays the others. Tests step it with
`compute.jobs['outbox-dispatch'].tick()`, which runs one pass over every sink.

<Mermaid
  chart="sequenceDiagram
  autonumber
  participant W as OutboxDispatch (the sink's lease holder)
  participant O as outbox
  participant S as your sink
  participant C as outbox_cursors
  W->>C: new lease epoch: load this sink's cursor
  loop each pass
    W->>O: read after the sink's cursor (finished transactions only)
    W->>S: deliver(batch)
    alt resolved
      W->>C: save the cursor, forward only
    else threw
      W->>W: bisect or back off; cursor unchanged
    end
  end"
/>

* **Per-sink cursors and leases.** Each sink reads from its own position under its own lease, so a failing or slow
  sink backs off without holding the others back.
* **Batches.** A pass reads up to 4 × `worker.batchSize` events per sink (128 with the default of 32) and hands them
  to `deliver` in one call.
* **Failure.** A throw leaves the cursor where it was. A failing batch is bisected: the next pass hands the sink
  half as many events, until the failing event is alone; the batch size grows back after successes. A lone event
  that fails is retried after 1 second, doubling up to 5 minutes, with a warning logged with the attempt count. See
  [poison events](#poison-events) for when an event is dead-lettered.
* **Handover.** Cursors are saved to `outbox_cursors` after every delivered batch. When a process takes a sink's
  lease under a new epoch it reloads that sink's cursor, so a new holder resumes where the last one saved. A batch delivered whose save
  failed, or whose holder died before saving, is delivered again.
* **A new sink** has no stored cursor and starts at the origin: it receives every event the outbox still holds.

## Cursors and the txid pitfall [#cursors-and-the-txid-pitfall]

```sql
CREATE TABLE outbox (
  seq bigserial PRIMARY KEY,
  txid xid8 NOT NULL DEFAULT pg_current_xact_id(),
  topic text NOT NULL,
  payload jsonb
);
CREATE TABLE outbox_cursors (            -- core/0009_runtime
  name text PRIMARY KEY,                 -- sink:<id>
  stream text NOT NULL DEFAULT 'outbox', -- or authz_changes
  txid xid8 NOT NULL,
  seq bigint NOT NULL,
  updated_at timestamptz NOT NULL DEFAULT compute_now()
);
```

An event's position is the pair `(txid, seq)`, never `seq` alone. Sequence numbers are drawn when a row is
inserted, but transactions commit in a different order: transaction A can take `seq` 41, transaction B `seq` 42,
and B commit first. A reader that saw 42 and moved its cursor past it would never see 41. Every reader in Better
Compute therefore reads like this:

```sql
SELECT o.seq::text AS seq, o.txid::text AS txid, o.topic, o.payload FROM outbox o
 WHERE o.txid < pg_snapshot_xmin(pg_current_snapshot())
   AND (o.txid, o.seq) > ($1::xid8, $2::bigint)
 ORDER BY o.txid, o.seq LIMIT $3;
```

Only transactions older than every transaction still running are read, so nothing can commit behind the cursor
afterwards. If you read the outbox yourself, keep all of it:

* **Filter on `txid < pg_snapshot_xmin(…)`.** Without it you will skip events under load, and no test on an idle
  database will show it.
* **Compare and order the pair.** `(txid, seq) > (…)` as a row comparison, `ORDER BY txid, seq`.
* **Order by the columns, not their text.** A query that selects `txid::text AS txid` and then says `ORDER BY txid,
  seq` sorts the text aliases (`'10' < '9'`). Qualify the columns (`o.txid`) or alias the text differently
  (`txid_text`), as the built-in readers do.
* **Keep the numbers as strings.** `xid8` is 64 bits; carry it as a decimal string and compare with `BigInt`, never
  as a JavaScript number.
* **Two spellings.** Server-side events (`events.list`, sinks, `compute.events`) write `sequence` as `<txid>:<seq>`.
  The events package's readers, the SSE stream and webhook bodies write `<txid>.<seq>`. They name the same position.

Cursor saves only move forward: the upsert applies only when the new `(txid, seq)` is greater, so a late save by a
holder that lost the lease cannot move a cursor back.

**Retention.** `retentionModule()` from `@better-compute/postgres` deletes outbox rows only behind the lowest cursor
of the `outbox` stream (and older than a minimum age); a stream with no cursor is never pruned. So a sink that stops
delivering holds the outbox back, and a sink you remove from the options leaves its `sink:<id>` row behind: delete
that row, or retention keeps waiting for it. On a schema without `outbox_cursors`, cursors live in memory, a
warning is logged once, and a new lease holder starts every sink at the origin.

## Poison events [#poison-events]

OutboxDispatch isolates an event your sink can never accept, so it no longer blocks the sink:

1. A batch that throws is bisected pass by pass until the failing event is delivered alone.
2. The lone event is retried with back-off. After 5 failures it is tested against its neighbour: the next event is
   delivered on its own.
3. If the neighbour goes through, the lone event is poison. It is written to `outbox_dead_letters` (the event as
   delivered, the error and the attempt count), logged as an error, and the sink's cursor moves past both events.
4. If the neighbour fails too, the sink itself is down: nothing is dead-lettered and the back-off goes on, so an
   outage never loses events. A failing event with no neighbour yet (the last one in the outbox) keeps retrying
   until one exists.

```sql
CREATE TABLE outbox_dead_letters (       -- core/0019_outbox_dead_letters
  sink text NOT NULL,                    -- EventSink.id
  txid xid8 NOT NULL,
  seq bigint NOT NULL,
  event jsonb NOT NULL,                  -- the ComputeEvent as delivered
  error text NOT NULL,
  attempts integer NOT NULL CHECK (attempts > 0),
  dead_at timestamptz NOT NULL DEFAULT compute_now(),
  PRIMARY KEY (sink, txid, seq)
);
```

On a schema without the table the event is only logged, and the cursor still moves past it. Nothing redelivers a
dead-lettered event: read the table and replay what you need yourself. Retention does not prune it.

The dispatcher tells "this event" from "this destination" only by whether the neighbour goes through, right after
the fifth failure. When your sink can tell a permanent failure from a transient one, it can do better itself:
isolate poison inside `deliver` and keep the dead letters where your destination lives.

```ts title="warehouse-sink.ts"
import { defineSink } from 'better-compute';

export const warehouse = defineSink({
  kind: 'events',
  id: 'warehouse',
  async deliver(events) {
    try {
      await store.insertMany(events); // the fast path: one idempotent write
      return;
    } catch (error) {
      if (isTransient(error)) throw error; // retried with back-off, cursor unchanged
    }
    // A permanent failure somewhere in the batch: store what you can, park the rest.
    for (const event of events) {
      try {
        await store.insertMany([event]);
      } catch (error) {
        if (isTransient(error)) throw error;
        await store.deadLetter(event, String(error)); // your own table or queue
      }
    }
  },
});
```

Throw only for failures that a retry can fix (the destination is down, a timeout). An event you record and skip
yourself never costs the sink five retries and a neighbour test.

## Webhooks are not a sink [#webhooks-are-not-a-sink]

Per-tenant webhooks are built differently, as the `eventsModule()` from `better-compute/events`, and they show the
other way to consume the outbox: from a module of your own.

|               | Event sink (`sinks.events`)                                                                                     | `eventsModule()` webhooks                                                                                                               |
| ------------- | --------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- |
| Runs in       | OutboxDispatch, the `sinks` role                                                                                | Reconcilers `events.fanout` (every 1 s), `events.deliver` (every 2 s) and `events.prune` (hourly), the `controllers` role               |
| Cursor        | `outbox_cursors`, `sink:<id>`; holds retention back                                                             | `event_cursors`, `webhooks`; does not hold retention back                                                                               |
| Starts at     | The origin                                                                                                      | The outbox tail when the module first starts: history is not replayed                                                                   |
| Unit of retry | A batch, bisected down to one event; no attempt limit while the neighbour fails too                             | One delivery row per webhook and event: backoff 1 m, 5 m, 30 m, 2 h, 6 h, 12 h, 24 h, then `dead` after 8 attempts                      |
| Poison        | Dead-lettered to `outbox_dead_letters` after 5 lone failures when its neighbour goes through; never redelivered | A failing endpoint dead-letters its own deliveries; `webhooks.redeliver` retries one; `events.prune` deletes dead letters after 90 days |

The fan-out pass locks its cursor row, reads a page with `readOutbox`, inserts one `webhook_deliveries` row per
matching webhook and event, and advances the cursor, all in one transaction. A module that needs per-item retries
can follow the same pattern with `readOutbox`, `outboxTail` and `formatCursor` from `better-compute/events`. Remember
that such a cursor is invisible to retention: if you need every event, save a copy of your position in
`outbox_cursors` too, or use a sink. See [webhooks](/docs/guides/events/webhooks).

For in-process reactions, `compute.events.subscribe('reservation.*', handler)` tails the outbox from the moment you
subscribe, with an in-memory cursor. It is not durable: use a sink when you need every event.

## Audit sinks [#audit-sinks]

`defineSink({ kind: 'audit', id, write })` builds an audit sink: `write(records: AuditRecord[]): Promise<void>`
receives committed rows of `audit_events`, for a SIEM or an archive. List it in `betterCompute({ sinks: { audit } })`
or in a module's or plugin's `extensions`.

```ts title="compute.ts"
import { betterCompute, defineSink } from 'better-compute';

const siem = defineSink({
  kind: 'audit',
  id: 'siem',
  async write(records) {
    // records: { id, tenantId, time (epoch ms), action, resource: { type, id }, admission?, details? }[]
    await siemClient.send(records.map((record) => ({ key: record.id, body: record })));
  },
});

export const compute = betterCompute({
  // …database, secret, baseURL, iam
  sinks: { audit: [siem] },
});
```

When at least one audit sink is listed, `betterCompute()` registers the built-in `audit-sinks` module (right after
the `tunnels` module, before your own). It works like OutboxDispatch, with its own tables:

|           | Audit sinks                                                                                                                      |
| --------- | -------------------------------------------------------------------------------------------------------------------------------- |
| Runs in   | The singleton reconciler `audit-sinks`, key `tick` every 5 s, the `controllers` role                                             |
| Reads     | `audit_events` after the sink's `(txid, seq)` cursor, finished transactions only (`txid < pg_snapshot_xmin(…)`), in commit order |
| Batches   | 500 records per `write`; a pass keeps going until a batch comes back short, so it drains a backlog                               |
| Cursor    | `audit_sink_cursors` (`sink_id`, `txid`, `seq`; migration `audit-sinks/0001_cursors`), saved forward only after each batch       |
| Failure   | The batch is retried after 1 s, doubling to at most 5 minutes, per sink; the other sinks go on. No bisecting and no dead letters |
| Guarantee | At least once: a batch written before its cursor was saved is written again. Deduplicate on `record.id`                          |

Ids must be non-empty and unique among the audit sinks (`INVALID_CONFIG` otherwise). A new sink starts at the origin
and receives every audit row still in the table. The `audit` module's hourly checkpoints still seal the rows in
place; a sink is a copy, not a replacement. See [audit](/docs/guides/events/audit).

## Usage [#usage]

`defineSink` also accepts `kind: 'usage'`, but a usage sink is refused: `sinks.usage` with any entry, or a usage sink
listed in a module's `extensions`, makes `betterCompute()` throw `INVALID_CONFIG`. Usage windows go to Better IAM
billing through the IAM bridge, which owns the single `recorded_at` flush. PLAN.md §5.4 sketches it as
`sinks: { usage: [bridge.usageSink()] }`; the code implements it as reconcilers:

1. `capacity.usage` (every 5 minutes) writes closed hourly usage windows of running
   allocations to `usage_windows`, each with the idempotency key `cmp:<allocationId>:<windowStartMs>`.
2. `iam.usage` in the bridge's `iam` module (every hour) sends up to 1,000 closed, unrecorded windows through
   `IamPort.recordUsage`, then stores each window's outcome.

`recordUsage` isolates poison the way an event sink should. Better IAM records a batch of up to 200 in one
transaction, so one bad event would fail the whole batch on every retry. On an error specific to the events
(`NOT_FOUND`, `CONFLICT`, `BILLING_PERIOD_CLOSED`, `INVALID_INPUT`, `INVALID_TRANSITION`, `METER_ARCHIVED`) the
batch is bisected down to the event, which gets one of these results:

| Result                  | Meaning                                                                                                                                     |
| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------- |
| `recorded`, `duplicate` | Recorded now, or already recorded under the same key                                                                                        |
| `reattributed`          | Its identity no longer exists: recorded under the offboarded identity's successor, or tenant-attributed with the tag `attribution=orphaned` |
| `dead-lettered`         | Still fails on its own; stored in `usage_dead_letters` with the error code and message                                                      |

Any other error is transient and rejects the whole call; the windows stay unrecorded and the next pass retries them
with the same keys. See [billing and usage](/docs/guides/iam/billing-and-usage).

## Checklist for your own sink [#checklist-for-your-own-sink]

1. Choose a stable `id`; it names the cursor.
2. Make `deliver` idempotent on `event.id`: every event can arrive twice.
3. Throw only for transient failures; record permanent ones yourself, or leave them to `outbox_dead_letters`.
4. Keep `deliver` within seconds: the sink's next batch waits for it (other sinks run in their own loops).
5. Expect a new sink to start with everything the outbox still holds.
6. When you retire a sink, delete its `sink:<id>` row from `outbox_cursors`.
7. Run at least one worker with the `sinks` role (`compute.start()` runs every role by default).