# Outbox adapters Kind reference For developer · operator Verified against @veritio/core@0.4.7 An outbox row is evidence that has already been committed but not yet delivered. Between that commit and the delivery sit the cases that break naive queues: two workers claiming the same row, a process that dies mid-attempt, a target that holds delivery indefinitely, and an outage that grows a backlog without limit. `@veritio/storage` exports one adapter interface, two storage backings, two dispatchers, one HTTP target, and a finite safety policy that bounds all of them. ## The module surface [Section titled “The module surface”](#the-module-surface) Export Module Role `createFileOutboxAdapter` `outbox-file.ts` Process-safe file queue with an optional backlog ceiling `OutboxQueueFullError` `outbox-file.ts` Typed refusal at that ceiling `createPostgresOutboxAdapter`, `createNeonOutboxAdapter`, `createMysqlOutboxAdapter`, `createMariaDbOutboxAdapter` `outbox-sql.ts` SQL queues that run inside the host’s transaction `POSTGRES_OUTBOX_SCHEMA_SQL`, `MYSQL_OUTBOX_SCHEMA_SQL` `outbox-types.ts` The DDL those adapters query against `dispatchOutboxEntry`, `createOutboxDispatcher` `outbox-dispatcher.ts` Delivery to a local `OutboxEvidenceTarget` `createHttpIngestTarget`, `createHttpOutboxDispatcher`, `DEFAULT_HTTP_INGEST_TIMEOUT_MS` `ingest-target.ts` Delivery over HTTP to a conforming ingest endpoint `DEFAULT_DELIVERY_SAFETY_POLICY`, `parseDeliverySafetyPolicy` `delivery-safety.ts` The finite ceilings every dispatch pass is checked against Everything here is TypeScript-only. The Python and Go SDKs produce the same governed drafts, but the queue machinery has no port yet. ## createFileOutboxAdapter [Section titled “createFileOutboxAdapter”](#createfileoutboxadapter) ```ts import { createFileOutboxAdapter } from '@veritio/storage' const adapter = createFileOutboxAdapter('./.veritio/outbox', { maxQueuedBytes: 5 * 1024 * 1024, }) ``` The adapter keeps one `entries.json` snapshot in `dir`. Every mutating operation runs inside `withLock`, which creates `.outbox.lock` with the `wx` flag, retries every 10 ms for up to 500 attempts, and removes the lock in a `finally`. Writes go to a uniquely named temporary file in the same directory and are published with `rename`, so a reader never observes a half-written snapshot and a crash mid-write leaves the previous complete one. `maxQueuedBytes` must be a positive safe integer; anything else throws a `TypeError` at construction. Omitting it means unbounded, which is the pre-existing behavior for host-managed queues. ## The transaction and enqueue contract [Section titled “The transaction and enqueue contract”](#the-transaction-and-enqueue-contract) `transaction` is the only write entry point on `OutboxAdapter`: ```ts await adapter.transaction(async (tx) => { await applyInvoiceMutation(after) await tx.enqueue({ id: draft.changeRef.id, tenantId: scope.tenantId, payload: draft.outboxEntry, }) }) ``` The file adapter stages enqueues against a cloned snapshot and writes only after the callback resolves, so a throwing callback discards them. The SQL adapters hand `transaction` straight to the injected client’s own `transaction`, which is what puts the application row and the queue row in a single commit. `OutboxEnqueueInput` has four fields: `id`, `tenantId`, `payload`, and an optional `availableAt` accepting a string or `Date`. `createPendingEntry` normalizes it into an `OutboxStoredEntry` with `status: "pending"`, `attempts: 0`, and ISO `createdAt` / `updatedAt` / `availableAt`, cloning the payload so later caller mutation cannot reach durable state. Enqueue is idempotent on `id`. Identity is compared as `tenantId` plus `canonicalJson(payload)` — not timestamps or attempt counters — so re-enqueueing the same id with the same payload returns the stored row, and re-enqueueing it with a different payload throws `outbox idempotency conflict` rather than overwriting a recorded intent. `validatePayload` runs on enqueue, on every read of a stored row, and again inside `dispatchOutboxEntry`. It requires `schemaVersion` to equal `2026-06-23`, `mutationBinding` to be one of `same_transaction`, `not_transaction_bound`, or `best_effort`, `records` and `edges` to be arrays, and every record and edge to carry a `scope.tenantId` matching the row’s tenant. A payload that mixes tenants is refused at the boundary, never partially delivered. Queue order is `createdAt` then `id`, compared as strings. Both are ISO-8601 UTC, so lexicographic order is chronological order, and the ordering survives restarts without depending on file names or insertion order. ## The SQL adapters and their DDL [Section titled “The SQL adapters and their DDL”](#the-sql-adapters-and-their-ddl) ```ts import { createPostgresOutboxAdapter, POSTGRES_OUTBOX_SCHEMA_SQL } from '@veritio/storage' const adapter = createPostgresOutboxAdapter({ client, tableName: 'app.veritio_outbox_entries' }) ``` `client` must satisfy `SqlOutboxExecutor`: an `execute(statement, params)` plus a `transaction(run)` that gives the callback a session on one real database transaction. The adapter issues parameterized statements only — `$n` for Postgres, `?` for MySQL — and `tableName` passes through `quoteTableName`, which accepts a plain or schema-qualified identifier matching `[A-Za-z_][A-Za-z0-9_]*` and rejects anything else before a statement is generated. `createNeonOutboxAdapter` is the Postgres dialect; `createMariaDbOutboxAdapter` (and the `createMariaDBOutboxAdapter` spelling) is the MySQL dialect. Run the exported DDL from your own migration tooling. The Postgres form: ```sql CREATE TABLE IF NOT EXISTS veritio_outbox_entries ( id text PRIMARY KEY, tenant_id text NOT NULL, payload_canonical text NOT NULL, entry_json text NOT NULL, status text NOT NULL, attempts integer NOT NULL, available_at text NOT NULL, created_at text NOT NULL, updated_at text NOT NULL, dispatched_at text, last_error text ); CREATE INDEX IF NOT EXISTS veritio_outbox_dispatch_idx ON veritio_outbox_entries (status, available_at, tenant_id, created_at, id); CREATE INDEX IF NOT EXISTS veritio_outbox_tenant_claim_idx ON veritio_outbox_entries (tenant_id, status, available_at, created_at, id); CREATE INDEX IF NOT EXISTS veritio_outbox_tenant_guard_idx ON veritio_outbox_entries (tenant_id, id); ``` `MYSQL_OUTBOX_SCHEMA_SQL` is the same shape with backtick quoting, `varchar` keys, `longtext` payload columns, and the three indexes declared as `KEY`s. Three details in that table are load-bearing. Timestamps are stored as text because the adapter compares normalized ISO strings, which keeps ordering and lease expiry identical across drivers that would otherwise apply their own time zone and precision handling. `entry_json` holds the complete serialized entry while `payload_canonical` holds only `canonicalJson(payload)`; every read recomputes the canonical form and throws `stored outbox entry integrity check failed` when the two disagree, so a row edited outside the adapter is caught before it can drive a delivery. The indexes exist because `claimDispatchable` filters on status, availability, and tenant simultaneously. Claiming is transactional. The adapter first locks a single stable tenant row (`ORDER BY id ASC LIMIT 1 FOR UPDATE`), then selects candidates with `FOR UPDATE SKIP LOCKED`, excluding any tenant that has a `paused` row or an unexpired `leased` row. The result is at most one active lease per tenant across every worker, and an expired lease becomes claimable again — that is the crash-recovery path, not an error path. `markDispatched` and `markFailed` both call `assertLease`, so a worker whose lease expired cannot settle a row another worker now owns. ## dispatchOutboxEntry and OutboxEvidenceTarget [Section titled “dispatchOutboxEntry and OutboxEvidenceTarget”](#dispatchoutboxentry-and-outboxevidencetarget) ```ts import { dispatchOutboxEntry, type OutboxEvidenceTarget } from '@veritio/storage' const target: OutboxEvidenceTarget = { recordEvent: (input) => store.append(createAuditEvent(input), { idempotencyKey: input.id }), recordEdge: (input) => edgeSink.append(createEvidenceEdge(input)), } await dispatchOutboxEntry(payload, target) ``` `recordEvent` takes an `AuditEventInput` and returns an `AuditRecord`; `recordEdge` takes an `EvidenceEdgeInput` and returns an `EvidenceEdgeRecord`. That is the entire interface — the queue never learns how the target stores anything. Ordering is fixed: ```text validatePayload(payload) → for each payload.records[i] → target.recordEvent (sequential) → for each payload.edges[i] → target.recordEdge (sequential) ``` Records first, because an edge is a relation between refs. Appending `change.declared`, `activity.recorded`, and `entity.revision.created` before anything claims a relation between them means a target that reads its own graph never observes an edge pointing at an event that does not exist yet. A target that fails halfway leaves a prefix appended and the row still undelivered. That prefix is safe only because idempotency is owed by the target, not provided by the queue. Every event and edge id in the payload is derived deterministically by the governed draft, so a sink keyed on those ids turns at-least-once delivery into exactly-once effect. A sink that appends blindly inflates the tenant chain on the first retry. ## Permits and the two dispatchers [Section titled “Permits and the two dispatchers”](#permits-and-the-two-dispatchers) `createOutboxDispatcher({ adapter, target, deliverySafety? })` drives a local target; `createHttpOutboxDispatcher({ adapter, target, deliverySafety? })` drives an HTTP one with a single POST per entry. Both expose `dispatchBatch(options)` and both require a `DispatchPermit`, checked by `validateDispatchOptions`: Rule Failure `kind` is `automatic`, `canary`, or `operator` `permit.kind is invalid` `maxEntries`, `maxBytes`, `maxElapsedMs`, `leaseMs` are positive safe integers `permit. must be a positive safe integer` `leaseMs` exceeds `maxElapsedMs` `permit.leaseMs must exceed permit.maxElapsedMs` an `operator` permit carries `approvalId` `permit.approvalId is required` `maxBytes` within `hard.rollingSendBytes`, `maxEntries` within `hard.rollingRequests`, `maxElapsedMs` within `hard.windowMs` exceeds the corresponding hard limit an `automatic` permit stays within `hard.automaticReplayBatches` `automatic permit exceeds the hard replay canary limit` A pass stops at the first of: `maxEntries` reached, `maxElapsedMs` elapsed, the rolling ledger out of requests or bytes, or no claimable entry. `createRollingWindowLedger` holds request and byte accounting per dispatcher instance across passes, so looping one-entry permits cannot walk past the window ceiling; bytes are reserved before the send, not after it. Its scope boundary is one process — the per-tenant single active lease is what serializes concurrent dispatchers. The two dispatchers differ on failure. The local one marks `retry` and breaks. The HTTP one honors the typed verdict: `retry` and `pause` break the pass, while `reject` dead-letters the row and continues, because a single malformed entry should not block the rest of the tenant’s backlog. ## createHttpIngestTarget [Section titled “createHttpIngestTarget”](#createhttpingesttarget) ```ts import { createHttpIngestTarget, DEFAULT_HTTP_INGEST_TIMEOUT_MS } from '@veritio/storage' const target = createHttpIngestTarget({ baseUrl: process.env.VERITIO_BASE_URL, // resolved at the process boundary key: process.env.VERITIO_INGEST_KEY, // never read inside the SDK timeoutMs: DEFAULT_HTTP_INGEST_TIMEOUT_MS, }) ``` `baseUrl` and `key` are required non-empty strings, `path` defaults to `/api/ingest`, `fetchImpl` defaults to the global `fetch` and throws if neither is a function, and `timeoutMs` defaults to `DEFAULT_HTTP_INGEST_TIMEOUT_MS`, which is `10_000`. `postBatch` sends `x-veritio-delivery: live-v1`; `dispatchEntry` sends `replay-v1`. An empty batch returns early with zeroed counts and no network call. Before sending, the encoded body is measured against `hard.batchBytes` and refused locally as a `413` `IngestClientError`. Every attempt is bounded because an unbounded one is indistinguishable from a hang. `AbortSignal.timeout(attemptTimeoutMs)` is attached to each request, `dispatchEntry`’s `timeoutMs` can only lower the configured value (`Math.min`), and `createHttpOutboxDispatcher` clamps each in-flight request to the permit time remaining. A slow target therefore cannot make a pass outlive its permit, and it cannot outlive the lease either — `leaseMs` must exceed `maxElapsedMs` — so a stalled worker’s row returns to the queue instead of wedging. Class Trigger Disposition `IngestRetryableError` `5xx`, or a body disposition of `retry` `retry` `IngestPausedError` body disposition `pause`, or a known legacy pause code `pause` `IngestConflictError` `409` with no explicit disposition `reject` `IngestRejectedError` explicit body disposition `reject` `reject` `IngestClientError` any other `4xx`, and the local `413` `reject` Disposition resolves in that order: a validated `deliveryDisposition` in the body wins outright, then a recognized legacy pause code, then the status-only fallback of `5xx` to retry and everything else to reject. Message text is written by the SDK from the status; server prose is never echoed. From a pause response only a `circuitId` matching `^[A-Za-z0-9_-]{1,128}$` and a positive integer `retryAfterSeconds` are retained, and `resolveCircuitId` substitutes a locally generated id when the server omitted one so the hold stays resumable. `resumePaused` then re-arms rows only when the operator supplies that exact `expectedCircuitId`. ## Delivery-safety policy [Section titled “Delivery-safety policy”](#delivery-safety-policy) `DEFAULT_DELIVERY_SAFETY_POLICY` is deep-frozen and conservative: Field Default `recommended.queuedBytes` 5 MiB `recommended.rollingSendBytes` 1 MiB `hard.batchBytes` 1 MiB `hard.queuedBytes` 64 MiB `hard.rollingSendBytes` 2 MiB `hard.rollingRequests` 30 `hard.windowMs` 60000 `hard.automaticReplayBatches` 1 `parseDeliverySafetyPolicy(undefined)` returns those defaults. Any other input must be a complete object with both branches: every threshold a positive safe integer, so zero can never be read as unlimited, and `automaticReplayBatches` exactly `0` or `1`, which caps unattended recovery at one canary batch. Three cross-checks then run — `recommended.queuedBytes` within `hard.queuedBytes`, `recommended.rollingSendBytes` within `hard.rollingSendBytes`, and `hard.batchBytes` within `hard.rollingSendBytes` — and the result is frozen. A partial policy fails closed rather than inheriting defaults for the fields it omitted. The backlog ceiling is enforced separately, by the file adapter’s own `maxQueuedBytes`; the policy’s `queuedBytes` fields are the thresholds a host maps onto that option, not a wire the adapter reads for itself. At the ceiling the behavior is deliberate: ```text backlog(pending + leased + paused + dead) + newEntryBytes > maxQueuedBytes → throw OutboxQueueFullError → every existing row stays exactly where it is ``` Existing evidence is retained and the newest enqueue is refused. Nothing is evicted to make room, because evicting the oldest row would silently delete the evidence the queue exists to protect. `dispatched` rows do not count toward the ceiling, and the error carries only the ceiling value, so it is safe to log. The SQL adapters do not implement this ceiling and rely on the host’s own database limits. ## What an entry carries [Section titled “What an entry carries”](#what-an-entry-carries) The checked roundtrip drafts an invoice change, commits it with the serialized entry, drains the queue, and dispatches the same stored row a second time: verified outbox roundtrip ```json { "draft": { "changeId": "chg_invoice_inv_123_bdb81c726b6eb5cc", "activityId": "act_invoice_inv_123_bdb81c726b6eb5cc", "entityId": "inv_123", "revisionId": "rev_invoice_inv_123_5bd31df6f81c_9d1ee3ed", "changedPaths": [ "/status" ], "stateDigest": "sha256:5bd31df6f81c5771276d1ad838a083f08715e90c1e4ca8d4faddae729280ed07", "stateCommitmentFields": { "amountCents": 48000, "currency": "USD", "customerEmail": { "captureMode": "content_digest", "digest": "sha256:e0bc5fb4b660ccb9376b5658539849d9d3e30f3cd147196de2ac6172f33e1013" }, "id": "inv_123", "status": "paid" } }, "outboxEntry": { "schemaVersion": "2026-06-23", "mutationBinding": "same_transaction", "recordCount": 3, "edgeCount": 4 }, "queueBefore": [ { "id": "outbox_chg_invoice_inv_123_bdb81c726b6eb5cc", "tenantId": "org_acme", "status": "pending", "payloadBytes": 5969 } ], "payloadContainsRawEmail": false, "appliedInvoiceStatus": "paid", "queueAfter": [ { "id": "outbox_chg_invoice_inv_123_bdb81c726b6eb5cc", "tenantId": "org_acme", "status": "dispatched" } ], "recordedEvents": [ { "sequence": 1, "action": "change.declared", "targetType": "change" }, { "sequence": 2, "action": "activity.recorded", "targetType": "activity" }, { "sequence": 3, "action": "entity.revision.created", "targetType": "invoice" } ], "recordedEdges": [ { "sequence": 1, "relation": "has_activity", "from": "change", "to": "activity" }, { "sequence": 2, "relation": "has_output", "from": "change", "to": "revision" }, { "sequence": 3, "relation": "performed_by", "from": "activity", "to": "principal" }, { "sequence": 4, "relation": "generated", "from": "activity", "to": "revision" } ], "replay": { "eventsAfterFirstDrain": 3, "eventsAfterReplay": 3, "edgesAfterFirstDrain": 4, "edgesAfterReplay": 4 }, "verification": { "audit": { "ok": true }, "edges": { "ok": true } } } ``` `payloadContainsRawEmail: false` is the minimization result: the email field was declared `content_digest` on the entity, so the durable row holds a digest. Minimization happens in the draft, before the payload is durable — the queue never sees the raw value and has nothing to strip at delivery time. `payloadBytes: 5969` is `outboxPayloadByteLength`, the exact UTF-8 size of the request body, which is the unit every ceiling in this page counts in. Three things are structurally absent from a row. The ingest key lives only in the target’s closure, injected at the host’s process boundary, and never touches the payload or the table. Server response text never lands in `lastError`, which stores an SDK-authored single-line summary truncated at 256 characters with no stack trace. And nothing but the governed draft’s own `records` and `edges` is stored, because `OutboxPayload` is typed directly as `GovernedChangeDraft["outboxEntry"]`. ## What a drained row proves [Section titled “What a drained row proves”](#what-a-drained-row-proves) `replay` shows three events and four edges after both drains: redelivery added nothing, and both chains still verify. That proves the durable intent survived storage as plain text and that the target’s idempotency held. It does not prove the host recorded every mutation that mattered, that the enqueue truly shared the application’s transaction — `mutationBinding` is a first-party assertion the SDK cannot observe — or that the row was delivered promptly. A row can sit `pending` for days and verify perfectly once drained; backlog age is an operational signal to monitor separately. Continue with the [transactional outbox guide](/docs/guides/transactional-outbox/) for the write path in context, [governed changes](/docs/concepts/governed-changes/) for the draft that produces each payload, and [storage overview](/docs/storage/overview/) to choose an authoritative target for the drain. [Edit page](https://github.com/getveritio/veritio-website/edit/main/src/content/docs/docs/storage/outbox.mdx) Last updated: Aug 23, 2026 [Previous File store](/docs/storage/file-store/)[Next Store conformance](/docs/storage/conformance/) Veritio provides evidence support, not legal advice or automatic compliance. This site uses cookieless, anonymous analytics (Umami) by default. With your consent, we also enable Google Analytics, which sets cookies and sends usage data to Google. [Privacy Policy](/legal/privacy/)