Sinks

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

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.

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

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.

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

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:

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

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.

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.

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

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.

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.

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

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.

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:

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.

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:

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.

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