This is the full developer documentation for QueueBox # QueueBox > A transactional outbox and an idempotent inbox as one service. Postgres or SQL Server. RabbitMQ, Kafka, NATS and HTTP. QueueBox is a self-hosted service. Your application writes a row in its own database transaction, and QueueBox delivers it. A broker or a webhook sends a message, and QueueBox stores it once, however often it arrives. ```sql BEGIN; INSERT INTO orders (id, customer_id, total) VALUES ('11111111-1111-1111-1111-111111111111', 'cust-42', 99.99); INSERT INTO outbox (topic, key, payload) VALUES ('order.created', 'cust-42', '{"orderId":"11111111-1111-1111-1111-111111111111"}'); COMMIT; ``` No lost messages The outbox row commits with your business write, or neither commits. QueueBox retries a failed delivery and moves an exhausted message to a dead-letter state that you can replay. No double processing The inbox stores a message once per source and idempotency key. A consumer claims one message of an aggregate at a time, in order. Order per key The outbox delivers the rows of one key in insert order, at any concurrency and with any number of replicas. Your database QueueBox uses Postgres or SQL Server that you already run. It needs no broker of its own, and it never reads your payload unless a transform asks it to. # Change data capture > Why QueueBox reads the database log, what capture changes and what it does not, and how one owner keeps its state. This page explains what change data capture does in QueueBox and why it cannot lose a message. For the setup steps, see [Capture changes](/how-to/capture-changes/). ## Why capture exists [Section titled “Why capture exists”](#why-capture-exists) Without capture, the outbox poller asks the database for due rows every `outbox.pollIntervalMs`. A short interval gives a short delay and a constant query load. A long interval gives a light load and a longer delay. Capture removes that trade. The capture connector reads the database log. When it sees an insert into the outbox table, it wakes the delivery loop at once. The delivery loop then claims and publishes through SQL, exactly as it does without capture. ## Capture only wakes delivery [Section titled “Capture only wakes delivery”](#capture-only-wakes-delivery) Capture delivers nothing. SQL stays the only source of truth: * The delivery loop claims every row with the same SQL claim as in polling mode. * A capture event that QueueBox never receives costs latency, not a message. * The reconciliation timer still wakes delivery at least once per `outbox.capture.reconciliationIntervalMs`. It also wakes delivery when the next retry or lease deadline falls due. * Delivery continues while capture is down, misconfigured or disabled. Capture ignores updates, deletes and tombstones. Only an insert or a snapshot record wakes delivery. A state change that QueueBox itself writes therefore creates no new work. ## The modes [Section titled “The modes”](#the-modes) | `outbox.capture.mode` | Database | Behaviour | | --------------------- | ---------- | ------------------------------------------------------- | | `polling` | Any | The default. Delivery polls at `outbox.pollIntervalMs`. | | `postgres-logical` | PostgreSQL | Delivery reacts to the logical replication stream. | | `sqlserver-cdc` | SQL Server | Delivery reacts to the change data capture tables. | The connector runs inside the QueueBox process. You operate no Kafka, no Kafka Connect and no Debezium Server. `enabled` is `false` and `mode` is `polling` by default. An upgrade therefore changes no behaviour until you ask for capture. Capture reads its host, port and database from `database.url`. The fields under `outbox.capture.connection` replace those values for the capture connection only. Use them to give capture a different account, a read replica, or one host when the URL lists several. ## The database side [Section titled “The database side”](#the-database-side) On PostgreSQL, capture reads a logical replication slot through a publication on the outbox table. You create the publication. QueueBox never creates or drops it, because an automatic publication can capture the wrong tables. The connector creates the replication slot on the first start, and QueueBox never drops it. Caution A replication slot holds write-ahead log until capture consumes it. A slot that nobody reads fills the disk of the database. Monitor `pg_replication_slots`, and drop the slot by hand when you retire a capture identity. On SQL Server, capture reads the change data capture tables of the outbox table. The SQL Server Agent must run. QueueBox checks that capture is enabled on the database and on the table before it starts the connector. ## The state directory [Section titled “The state directory”](#the-state-directory) The connector records how far it has read. `outbox.capture.stateDirectory` must be a durable volume that survives a restart. | File | Content | | ------------- | ------------------------------------------------------------------ | | `offsets.dat` | The log position that capture already delivered. | | `history.dat` | The schema history. SQL Server only. | | `identity` | The identifier that ties these files to the database registry row. | The image runs as the non-root user `queuebox` and ships `/var/lib/queuebox/capture` owned by that user. An empty Docker named volume mounted there inherits that ownership. A bind mount keeps the ownership of the host directory, so change its owner before you start QueueBox. A container without a mounted volume loses the files on every restart. ## Reconciliation of the state [Section titled “Reconciliation of the state”](#reconciliation-of-the-state) The table `queuebox_capture_state` records the capture identity and a fingerprint of the capture settings. QueueBox compares the table, the state files and the settings at start. The comparison detects three faults that otherwise pass unnoticed: * The volume is missing or empty, but the database says that capture ran before. * The state files belong to a different instance. * The settings changed. A new slot, publication, schema, table, host or database makes the recorded offsets meaningless. QueueBox also stops capture when the PostgreSQL replication slot disappeared but the state remains. Otherwise the connector creates a new slot and restarts from the present. A fault that needs a decision does not retry. QueueBox stops capture, reports the reason, and keeps delivering through SQL. Recovery is an operator decision, because every automatic answer either replays or drops log positions. [Capture changes](/how-to/capture-changes/) holds the recovery steps. A fresh snapshot after recovery delivers no message twice, because delivery claims each row through SQL. ## One owner [Section titled “One owner”](#one-owner) Exactly one process owns a capture identity. The owner holds a database session lock for the whole run. A second process that starts with the same identity fails the lock. It reports that the identity already has an owner, and it keeps delivering through SQL. Set `outbox.capture.enabled: false` on every replica that must not own capture. QueueBox does not elect an owner, so failover is manual. To move capture, stop the current owner, move or recreate the state directory, and start the new owner with the same identity. ## Health [Section titled “Health”](#health) Capture health is separate from delivery health. `/health/ready` reports the component `outbox-capture`, but that component is advisory. A capture fault never makes the readiness answer unhealthy, because the instance still delivers. When the connector fails, QueueBox: * marks capture unhealthy and keeps SQL delivery running, * retries with a backoff from one second up to thirty seconds, * wakes delivery on every attempt, so nothing waits for capture to recover. Watch the `outbox-capture` component to see a capture fault. Watch the delivery metrics to see whether messages move. # Claims and leases > How QueueBox replicas and pull workers take a message, keep it, lose it, and recover it after a crash. This page explains how a worker takes ownership of a row, and what happens when it loses that ownership. The same mechanism protects the outbox poller, the inbox relay and pull workers. ## The claim [Section titled “The claim”](#the-claim) A worker never works on a row that it has not claimed. A claim is one `UPDATE` that sets these columns: | Column | Value | | ------------------ | ------------------------------------------- | | `state` | `processing` | | `claim_token` | A new random UUID. | | `lease_expires_at` | The database clock plus the lease duration. | | `claimed_at` | The time of the claim. | The claim token names the owner. Only the worker that holds the token can complete the row. A new claim of the same row issues a new token, so an old owner cannot act on the row again. The lease is the time for which the claim is valid. The database clock computes it, so a clock difference between the application hosts and the database does not change it. | Worker | Lease duration | Default | | ------------- | ------------------------------------- | -------------- | | Outbox poller | `outbox.claimTimeoutMs` | 300000 ms | | Inbox relay | `inbox.relay.claimTimeoutMs` | 300000 ms | | Pull worker | The `lease_ms` that the client passes | Set per client | ## Renewal [Section titled “Renewal”](#renewal) A worker renews its lease every third of the lease duration while it works on the row. A renewal sets `lease_expires_at` to the database clock plus the lease duration again. A long publish or a long handler therefore keeps its claim. A renewal is fenced like every other write, as the next section describes. A renewal that updates zero rows, or that fails, means that the worker lost the claim. The worker then cancels its work on that row. ## The claim fence [Section titled “The claim fence”](#the-claim-fence) Every write that finishes a claim is fenced. This covers the write that completes the row, the retry, the dead-letter write and the renewal. The `UPDATE` matches all of these conditions: * the row `id`, * the state `processing`, * the `claim_token` of the caller, * a `lease_expires_at` later than the database clock. The write reports whether it changed a row. A write that changed zero rows lost the claim, and the worker changes nothing more. A pull worker follows the same rule. A renewal, completion, retry or dead-letter statement must affect exactly one row. Zero rows means the worker lost ownership: stop work, and never retry with a different token. ## Reclaim of stale claims [Section titled “Reclaim of stale claims”](#reclaim-of-stale-claims) A worker can crash, pause or lose its database connection while it holds a claim. Its lease then expires, because nothing renews it. * The outbox poller and the inbox relay run a reclaim step at most once per `claimTimeoutMs / 5`. The step returns every row in state `processing` whose lease expired to state `pending`. The next claim takes the row again with a new token. * A pull claim needs no separate step. The pull claim statement takes a `pending` row or a `processing` row whose lease expired, in one statement. The reclaim step counts outbox rows in `queuebox_outbox_messages_reclaimed_total`. A shutdown uses the same path. QueueBox waits up to `outbox.shutdownTimeoutMs` for in-flight messages. A message that is still in flight after that time stays in state `processing`, and the reclaim step recovers it after its lease expires. ## Lost claims and duplicates [Section titled “Lost claims and duplicates”](#lost-claims-and-duplicates) The reclaim step runs on a timer. It does not prove that the old owner died. A slow worker can outlive its own lease while another replica claims the same row. The claim fence decides what happens next. **Outbox poller.** The poller publishes first and marks `sent` after. When the mark loses the claim, the destination already holds the message, and the new owner publishes it again. QueueBox cannot undo a delivery. It logs an error, increments `queuebox_claims_lost_total{component="outbox"}`, and leaves the row to the new owner. A lost retry or a lost mark `dead` logs a warning and increments the same counter. The destination receives a duplicate, and an [idempotent receiver](/concepts/delivery-semantics/#an-idempotent-receiver) absorbs it. **Inbox relay.** The relay writes the outbox row and marks the inbox row `processed` in one transaction. When the mark loses the claim, the relay rolls the outbox insert back. The row therefore reaches the outbox once, and no second outbox row with a new `X-Message-Id` exists. The relay logs the loss and increments `queuebox_claims_lost_total{component="inbox"}`. **Pull worker.** Complete the row in the transaction of your business change. When the completion affects zero rows, roll back the whole transaction. External work outside that transaction can happen more than once. Deduplicate it on `(source, idempotency_key)`. Tip A moving `queuebox_claims_lost_total` means that work outlives its lease. Raise `outbox.claimTimeoutMs` or `inbox.relay.claimTimeoutMs` above the slowest publish or forward. ## The claim locks [Section titled “The claim locks”](#the-claim-locks) Row locks keep two concurrent claims off the same row. Some claims also take a lock that serializes the whole claim statement. | Claim | PostgreSQL | SQL Server | | ------------- | ------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------- | | Outbox poller | `FOR UPDATE SKIP LOCKED` on the rows. No claim lock. | An exclusive application lock `queuebox_outbox_claim_`, plus `UPDLOCK, READPAST, ROWLOCK`. | | Inbox relay | A transaction advisory lock (`pg_advisory_xact_lock`) keyed on the table name, plus `FOR UPDATE SKIP LOCKED`. | An exclusive application lock `queuebox_inbox_claim_`, plus `UPDLOCK, READPAST, ROWLOCK`. | | Pull worker | `FOR UPDATE SKIP LOCKED`. No claim lock. | An exclusive application lock on the source name, owned by the claim transaction. | The relay lock closes the window between two concurrent claims of one aggregate. Without it, two replicas can each see no `processing` row for an aggregate and each claim one. The lock is released when the claim transaction commits, so the relay holds it only for the claim, not for the forward. A SQL Server application lock waits up to 10 seconds. The relay and outbox locks belong to the database session and are released after the claim. The pull lock belongs to the claim transaction. See [Ordering](/concepts/ordering/#the-sql-server-pull-claim-serializes-per-source) for its throughput limit and its driver timeout rule. ## Why a claim takes one row of a key or aggregate [Section titled “Why a claim takes one row of a key or aggregate”](#why-a-claim-takes-one-row-of-a-key-or-aggregate) A claim returns at most one row of an outbox key, and at most one row of an inbox aggregate. Another row of that key or aggregate stays unclaimed until the first row is finished. The claim is therefore the place where QueueBox enforces [order](/concepts/ordering/). The reclaim and the fence keep that order across a crash: an old owner cannot finish a row that a new owner took. # Delivery semantics > What QueueBox promises for an accepted message, and what your application must do to hold its side of the contract. This page states what QueueBox promises for a message that it accepted. It also states what your application and your receivers must do in return. For the order of messages, see [Ordering](/concepts/ordering/). ## At least once [Section titled “At least once”](#at-least-once) QueueBox delivers every accepted message at least once. It never delivers a message zero times, and it can deliver a message more than once. * The outbox row commits with your business write, so a committed business change always has its message. * QueueBox retries a failed delivery until the row reaches its `max_attempts`, and then marks the row `dead`. A dead row stays in the table until an operator replays or deletes it, or retention removes it. * A crash between the publish and the mark `sent` delivers the message a second time. Every consumer must therefore tolerate a repeat. The inbox is how a QueueBox consumer does that. ## Push and pull [Section titled “Push and pull”](#push-and-pull) Every source declares how its messages leave the inbox. ```yaml # fragment sources: orders: type: http path: /orders idempotencyKeyPath: $.id consumption: pull # default: push ``` QueueBox stores the value on the row at receipt. A later change of the configuration never alters a message that the inbox already holds. | | `push` | `pull` | | ------------------------- | ------------------------------------------ | ---------------------------------- | | Who moves the message | The inbox relay | Your worker | | Needs a topic and a route | Yes | No | | The relay claims the row | Yes | Never | | `processed` means | QueueBox forwarded the row into the outbox | Your application finished the work | The relay claims push rows only. A pull row is invisible to it, so a pull source needs no topic, no route and no destination. `processed` carries two different meanings. For a push row, it says that the message reached the outbox. Delivery to the destination is a separate outbox state. For a pull row, it says that your application completed the business work. A pull worker claims rows, renews its lease, and completes, retries or dead-letters each row. A client library does this work for you, so your application writes only its handler. | Language | Package | | ---------- | ------------------------------------------- | | C# | `QueueBox.Inbox` on NuGet | | TypeScript | `@alternayte/queuebox-inbox` on npm | | Go | `github.com/alternayte/queuebox/clients/go` | See [Pull clients](/reference/pull-clients/) and [Consume the inbox](/how-to/consume-the-inbox/). ## Where a message comes from, and where it goes [Section titled “Where a message comes from, and where it goes”](#where-a-message-comes-from-and-where-it-goes) | Broker | As a source | As a destination | | -------- | ---------------------------------- | --------------------------------- | | HTTP | Yes, `POST /inbox/` | Yes | | RabbitMQ | Yes, one queue per source | Yes, one exchange per destination | | Kafka | Yes, one consumer group per source | Yes, one topic per destination | | NATS | Yes, JetStream only | Yes, JetStream by default | Each source keeps the promise of the broker that it reads. QueueBox acknowledges a message only after the inbox row commits. A crash therefore replays the message instead of losing it, and the unique constraint on `(source, idempotency_key)` rejects the replay. * **Kafka** commits the offset after the store. An offset says that everything before it is done. The consumer therefore commits only the unbroken run of records that the inbox accepted, and it seeks back to the first record that failed. * **NATS** acknowledges the message after the store. It negatively acknowledges a failed store, so JetStream returns the message at once. The source reads JetStream only, because core NATS cannot acknowledge, and an inbox on it loses every message that arrives during a restart. * **RabbitMQ** acknowledges after the store and requeues on failure. QueueBox stores a broker message whose body is not JSON in state `dead`, and acknowledges it. Nothing downstream can read such a body. A refusal to acknowledge it stops the partition or returns the message for ever. ## The identity of a message [Section titled “The identity of a message”](#the-identity-of-a-message) Three identifiers travel with a forwarded message. Each answers a different question. | Identifier | Where | Stable across | | --------------------------- | ------------------------------------------ | -------------------------------------- | | `(source, idempotency_key)` | The inbox unique constraint | Every retry and every replay | | `x-inbox-id` | An outbox header | The inbox row | | `X-Message-Id` | An outbox header, equal to the outbox `id` | One outbox row, every delivery attempt | The replay identity `(source, idempotency_key)` includes the source on purpose. Two sources can send the same event ID and mean different events. The relay copies the inbox `headers` onto the outbox row. Then it sets `x-inbox-id`, `x-source`, `x-idempotency-key` and `X-Correlation-Id`. Each of these four replaces a received header of the same name in any letter case, so a sender cannot set them. See [Headers](/reference/headers/). A replay of an inbox row creates a **new** outbox row with a new `X-Message-Id`. A receiver that deduplicates on `X-Message-Id` alone accepts the replay as new work. Deduplicate relay traffic on `x-idempotency-key`, which carries the inbox idempotency key unchanged. ### Where the idempotency key comes from [Section titled “Where the idempotency key comes from”](#where-the-idempotency-key-comes-from) An HTTP source reads the key with `idempotencyKeyPath`. A broker source reads the `x-idempotency-key` header first, then `idempotencyKeyPath`, then the AMQP `messageId` property. When all of them give nothing, QueueBox uses a SHA-256 digest of the body. Caution The digest treats identical bytes as one message. Two distinct events with an identical body deduplicate to one row, and QueueBox does not forward the second event. Configure a key source when two events can carry the same body. ### Join the inbox and the outbox [Section titled “Join the inbox and the outbox”](#join-the-inbox-and-the-outbox) Join the two tables through the header when you investigate a message. * PostgreSQL ```sql SELECT o.id, o.topic, o.state, o.attempt, i.source, i.idempotency_key FROM outbox o JOIN inbox i ON i.id = (o.headers ->> 'x-inbox-id')::uuid WHERE i.source = 'stripe' AND i.idempotency_key = 'evt_123'; ``` * SQL Server ```sql SELECT o.id, o.topic, o.state, o.attempt, i.source, i.idempotency_key FROM outbox o JOIN inbox i ON i.id = CAST(JSON_VALUE(o.headers, '$."x-inbox-id"') AS UNIQUEIDENTIFIER) WHERE i.source = N'stripe' AND i.idempotency_key = N'evt_123'; ``` ## HTTP delivery [Section titled “HTTP delivery”](#http-delivery) QueueBox sends the payload as the body of a POST, with the headers in [Headers](/reference/headers/). Every 2xx status completes the delivery. No other status completes it. QueueBox never follows a redirect, so a 3xx answer fails the attempt. ### A 202 transfers durable responsibility [Section titled “A 202 transfers durable responsibility”](#a-202-transfers-durable-responsibility) A `202 Accepted` completes the delivery exactly as a `200 OK` does. QueueBox marks the outbox row `sent` and never sends that row again. Answer 202 only after the message is durable at the receiver. A receiver that answers 202 and then loses the message in memory loses it for good. QueueBox holds no copy that it retries, because a 2xx says that the receiver took responsibility. A receiver that cannot store the message yet must answer a status that is not 2xx, so that QueueBox retries. ### Duplicates [Section titled “Duplicates”](#duplicates) A receiver must be idempotent. Two situations create a duplicate even when nothing is broken. * **The response is lost.** The receiver stored the message and answered, the answer never arrived, and QueueBox retries the same `X-Message-Id`. * **The claim expires during a slow publish.** QueueBox publishes, another replica takes over the row, and that replica publishes it again. QueueBox logs the duplicate and counts it in `queuebox_claims_lost_total{component="outbox"}`. Raise `outbox.claimTimeoutMs` above the slowest publish to remove the cause. QueueBox never drops a duplicate in silence, and it never rolls a delivery back. See [Claims and leases](/concepts/claims-and-leases/#lost-claims-and-duplicates). ### An idempotent receiver [Section titled “An idempotent receiver”](#an-idempotent-receiver) Store the identifier and the effect in one transaction. A repeat then finds the row and changes nothing, whatever the reason for the repeat. * PostgreSQL ```sql BEGIN; INSERT INTO delivery_receipts (message_id, idempotency_key, received_at) VALUES ($1, $2, now()) ON CONFLICT (idempotency_key) DO NOTHING; -- Zero rows means this is a repeat. Skip the business work and answer 200. -- One row means this is new work. Apply the business change here, in this -- transaction, and answer 200 only after the commit succeeds. COMMIT; ``` * SQL Server ```sql BEGIN TRANSACTION; INSERT INTO delivery_receipts (message_id, idempotency_key, received_at) SELECT @message_id, @idempotency_key, SYSUTCDATETIME() WHERE NOT EXISTS ( SELECT 1 FROM delivery_receipts WITH (UPDLOCK, HOLDLOCK) WHERE idempotency_key = @idempotency_key ); -- Zero rows means this is a repeat. One row means new work: apply it here. COMMIT TRANSACTION; ``` Use `x-idempotency-key` as the key for relay traffic. Use `X-Message-Id` for traffic that a producer wrote directly into the outbox. Answer a 2xx only after the transaction commits. A 2xx before the commit turns a receiver crash into a lost message, because QueueBox does not send the message again. ## Retention [Section titled “Retention”](#retention) Retention is off by default. Turn it on for each table separately in the [configuration](/reference/configuration/). Retention never deletes active work. In the outbox, it deletes rows in state `sent` or `dead`. In the inbox, it deletes rows in state `processed` or `dead`. A row in state `pending` or `processing` stays, whatever its age. Deleting an inbox row ends deduplication for that message. The unique constraint on `(source, idempotency_key)` rejects a repeat, and a deleted row no longer rejects anything. Set the inbox retention age above the longest window in which a sender can repeat a delivery. If you do not, QueueBox accepts the repeat as a new message. # How QueueBox works > The modules, the path of a message through the outbox and the inbox, the state sets, and the relay. This page explains the parts of QueueBox and the path that a message takes through them. It also holds the one authoritative list of the outbox and inbox states. ## Two tables and three workers [Section titled “Two tables and three workers”](#two-tables-and-three-workers) QueueBox is one process that works on two tables in your database. * The **outbox** table holds messages that your application wants to send. Your application inserts a row in the same transaction as its business write. The **outbox poller** claims the row and delivers it to a destination. * The **inbox** table holds messages that arrive from outside. An HTTP source or a broker consumer stores each message once per source and idempotency key. * The **relay** moves a stored inbox row into the outbox, so the outbox delivers it. The relay handles push sources only. A pull source leaves the row in the inbox for your own worker. A destination is one configured egress: an HTTP endpoint, a RabbitMQ exchange, a Kafka topic or a NATS subject. An inbox source is one configured ingress: a webhook path or a broker consumer. A route matches a topic pattern to a destination. QueueBox runs on PostgreSQL or SQL Server. Every replica reads and writes the same two tables. The database is the only coordination point between replicas: row locks, claim tokens and leases keep two replicas off the same message. See [Claims and leases](/concepts/claims-and-leases/). ## The modules [Section titled “The modules”](#the-modules) The source tree has one Gradle module per responsibility. | Module | Responsibility | | ---------------- | ----------------------------------------------------------------------------------------- | | `app` | The main application and the HTTP server: inbox routes, health, metrics and admin routes. | | `core` | The domain model and the repository interfaces. It depends on no other module. | | `config` | Configuration loading from YAML and `QUEUEBOX_` environment variables, and validation. | | `outbox-service` | The outbox poller, the router, the transforms and the HTTP publisher. | | `inbox-service` | The inbox routes, source authentication and the relay. | | `postgres` | The PostgreSQL repositories and migrations. | | `sqlserver` | The SQL Server repositories and migrations. | | `rabbitmq` | The RabbitMQ consumer and publisher. | | `kafka`, `nats` | The Kafka and NATS consumers and publishers. | | `capture` | Embedded change data capture. It only wakes delivery. | The repository layer loads a database provider by reflection. The `app` module ships both providers, so the published image runs on either database. The build fails when a provider leaves the runtime class path, because reflection hides that fault from the compiler. ## The path of an outbox message [Section titled “The path of an outbox message”](#the-path-of-an-outbox-message) 1. Your application inserts a row into `outbox` in state `pending`, in the transaction of its business write. The row commits with the business change, or neither commits. 2. The poller claims a batch of due rows. A claim moves each row to state `processing` and gives it a claim token and a lease. 3. The router matches the row `topic` against the route patterns. The first matching route names the destination. 4. The route transform and then the destination transform run, if configured. 5. The publisher sends the message to the destination. 6. On success, the poller marks the row `sent`. 7. On failure, the poller increments `attempt` and schedules a retry with exponential backoff. The row goes back to `pending` with a later `scheduled_at`. 8. When `attempt` reaches the `max_attempts` of the row, the poller marks the row `dead`. A row with no matching route, or with a destination that no publisher supports, goes to `dead` at once. A claim that a crash leaves behind returns to `pending` when its lease expires. The outbox delivers the rows of one `key` in insert order. See [Ordering](/concepts/ordering/#order-and-the-key). ## The path of an inbox message [Section titled “The path of an inbox message”](#the-path-of-an-inbox-message) 1. **Receive.** An HTTP source or a broker source accepts the message. An HTTP source answers at the path that `inbox.basePath` and the `path` of the source build, for example `/inbox/stripe`. 2. **Extract.** QueueBox reads the idempotency key, the event type and the aggregate identifier. 3. **Transform.** The source transform reshapes the payload, if one is configured. The transform runs before the duplicate check. A payload that the transform rejects never becomes a stored row. 4. **Store.** QueueBox writes the inbox row in state `pending`. The unique index on `(source, idempotency_key)` rejects a repeat. An HTTP source answers 202 for a new message and 200 with `{"status":"duplicate"}` for a repeat. 5. **Forward.** For a push source, the relay claims the row, writes an outbox row from it, and marks the inbox row `processed`. Both writes run in one transaction. 6. **Route and deliver.** The outbox poller routes and delivers the new outbox row as in the previous section. A 202 means that QueueBox stored the message. It does not mean that QueueBox delivered it. For a pull source, the path stops at step 4. Your worker claims the row, does the work, and marks it `processed` itself. See [Delivery semantics](/concepts/delivery-semantics/#push-and-pull). ## The relay [Section titled “The relay”](#the-relay) The relay forwards an inbox row into the outbox. It never reads the meaning of the payload, and it runs no transform: the transform ran at ingestion. The outbox machinery routes and delivers the forwarded row. The relay maps the inbox row to the outbox row as follows. | Outbox field | Value | | -------------- | ----------------------------------------------------------------------------------------------------------------------------- | | `topic` | The rendered `sources..topic` template. | | `key` | The inbox `aggregate_id`. | | `payload` | The stored inbox payload. | | `headers` | The inbox `headers`, then `x-inbox-id`, `x-source`, `x-idempotency-key`, and `X-Correlation-Id` when the message carries one. | | `max_attempts` | `inbox.relay.maxAttempts`, or `outbox.maxAttempts` when that is not set. | The four relay headers replace a received header of the same name in any letter case. A sender therefore cannot set them. The topic template accepts `{{ source }}` and `{{ eventType }}`. The default for an HTTP source is `{{ eventType }}`. The default for an AMQP source is `{{ source }}`, because an AMQP message carries no event type of its own. QueueBox refuses to start when an AMQP template uses `{{ eventType }}` and the source sets neither `eventTypePath` nor `eventTypeFromHeader: true`. The relay marks the inbox row `dead` when the template renders empty, because such a message can reach no destination. ```yaml # fragment inbox: relay: enabled: true pollIntervalMs: 100 batchSize: 100 claimTimeoutMs: 300000 ``` Set `inbox.relay.enabled: false` to turn the relay off. The inbox then becomes a log that only pull workers read. ## Outbox states [Section titled “Outbox states”](#outbox-states) The outbox state set: ```text pending processing sent dead ``` | From | To | When | | ------------ | ------------ | --------------------------------------------------------------------------------- | | (insert) | `pending` | Your application inserts the row. | | `pending` | `processing` | The poller claims the row. | | `processing` | `sent` | The destination accepts the message. | | `processing` | `pending` | A retry is scheduled, or the lease expired and the reclaim step returned the row. | | `processing` | `dead` | No attempt remains, no route matches, or a transform dead-letters the message. | `sent` and `dead` are final. Only an operator moves a row out of them, with a [replay](/how-to/replay-dead-letters/) or with [SQL](/operations/dead-letters/). ## Inbox states [Section titled “Inbox states”](#inbox-states) The inbox state set: ```text pending processing processed dead ``` | From | To | When | | ------------ | ------------ | -------------------------------------------------------------------------------------------------------- | | (store) | `pending` | A source accepts the message. | | (store) | `dead` | An AMQP source stores a message that its transform rejected, or a broker message whose body is not JSON. | | `pending` | `processing` | The relay or a pull worker claims the row. | | `processing` | `processed` | The relay forwarded the row, or a pull worker completed it. | | `processing` | `pending` | A pull worker schedules a retry, or the relay lease expired and the reclaim step returned the row. | | `processing` | `processing` | A pull claim takes over a row whose lease expired, with a new claim token. | | `processing` | `dead` | The topic template rendered empty, or a pull worker gave up. | `processed` means two different things. For a push row, the message reached the outbox, and the outbox state tracks the delivery. For a pull row, your application finished the work. An AMQP source writes a rejected message in state `dead` in one transaction, and only then acknowledges the broker delivery. The row never exists in state `pending`, so the relay cannot forward a payload that the transform rejected. QueueBox declares no dead-letter exchange, so that row is the only copy. The relay claims `pending` rows only, so no read path of QueueBox returns a dead inbox row. The row exists for an operator to read and requeue. See [Dead letters](/operations/dead-letters/). ## The state column [Section titled “The state column”](#the-state-column) Both schemas declare the state column 50 characters wide. PostgreSQL uses `VARCHAR(50)`, and SQL Server uses `NVARCHAR(50)`. Note In memory, `MessageState` in `core` represents a state. `MessageState.Sent` carries both the outbox literal `sent` and the inbox literal `processed`. `MessageState.Failed` is never written to the database: a repository returns it for a literal that it does not know. # Ordering > The order QueueBox keeps within one inbox aggregate and one outbox key, and the order it does not keep. This page states which messages QueueBox delivers in order and which it does not. The inbox orders by `aggregate_id`. The outbox orders by `key`. Nothing orders two different aggregates or two different keys. ## Order and the aggregate [Section titled “Order and the aggregate”](#order-and-the-aggregate) The inbox `aggregate_id` column is the unit of order on the inbox side. Both consumption modes hold one message of an aggregate in flight at a time. The two modes scope that reservation differently, so do not treat them as the same rule. * In push mode, the relay reserves one in-flight message per `aggregate_id` across every source together. The relay is one process that reads every push row, so its claim has no source term. * The relay forwards the messages of one aggregate in the order of `created_at`. * In pull mode, the claim reserves one in-flight message per `(source, aggregate_id)`, because a pull worker binds to one source. The claim never returns a message whose aggregate already holds a message in state `processing` under a live lease on that source. The rule holds across every worker instance of that source, because it lives in the claim statement. * A pull claim on one source never blocks a pull claim of the same aggregate on a different source. A message of aggregate `A` on source `orders` and a message of aggregate `A` on source `payments` can run at the same time. * QueueBox does not serialize the claim of two different aggregates. * A row with no `aggregate_id` takes part in no ordering. Set the aggregate identifier with `aggregateIdPath` on the source. A broker source can also read it from the `x-aggregate-id` header. ```yaml # fragment sources: orders: type: http path: /orders idempotencyKeyPath: $.eventId eventTypePath: $.type aggregateIdPath: $.orderId ``` ### How the relay keeps the aggregate order [Section titled “How the relay keeps the aggregate order”](#how-the-relay-keeps-the-aggregate-order) * The claim excludes an aggregate that already has a message in state `processing`. * The relay claims and forwards one message at a time. The outbox poller publishes with the configured concurrency, so parallel work happens at delivery, not at forwarding. * A database lock serializes the relay claim across every replica. See [Claims and leases](/concepts/claims-and-leases/#the-claim-locks). * The claim fence stops a replica that lost its claim from forwarding a second copy. Two replicas therefore never write one aggregate into the outbox twice. **Guarantee.** At most one message per aggregate is in state `processing` at any time, across every replica. The relay forwards the messages of one aggregate in creation order. The relay writes the `aggregate_id` into the outbox `key`. The outbox then delivers the rows of one key in insert order. The order of one aggregate therefore holds from the inbox to the destination. ## Order and the key [Section titled “Order and the key”](#order-and-the-key) The outbox `key` column is the unit of order on the outbox side. It is the inbox aggregate rule, applied to the delivery to a destination. * The outbox delivers the rows of one key in insert order, at any `outbox.concurrency` and with any number of replicas. * The database fills the `sequence` column on insert. The claim orders the rows of one key by `sequence`, not by `created_at`, so rows that one transaction writes keep their insert order. * The claim takes a row of a key only when no earlier row of that key is `pending` or `processing`. One row of a key is in flight at a time. * A row of a key that waits for a retry holds back the later rows of its key until the row is sent or dead. * A dead row releases its key, and the poller delivers the next row of the key. One poison row does not stall a stream. * A replay of a dead row delivers it after the rows that passed it. * A row with a null or empty `key` takes part in no ordering. The poller publishes it in parallel with other rows. * QueueBox does not order two different keys. ```sql BEGIN; INSERT INTO outbox (topic, key, payload) VALUES ('order.created', 'order-42', '{"step":1}'); INSERT INTO outbox (topic, key, payload) VALUES ('order.paid', 'order-42', '{"step":2}'); INSERT INTO outbox (topic, key, payload) VALUES ('order.shipped', 'order-42', '{"step":3}'); COMMIT; ``` The three rows above reach the destination in the order 1, 2, 3. ### One writer per key [Section titled “One writer per key”](#one-writer-per-key) The rule holds for rows that one writer inserts per key. Two transactions that insert rows of the same key at the same time can commit out of `sequence` order. A row that commits late can then arrive after a row with a higher `sequence`. An event store that checks the stream version on append has one writer per stream, so the rule holds for it. ### Throughput of one key [Section titled “Throughput of one key”](#throughput-of-one-key) One row of a key in flight limits the throughput of that key to one publish round trip per row. Spread a busy stream over more keys only when the consumer does not need order across them. ## Claim order, not commit order [Section titled “Claim order, not commit order”](#claim-order-not-commit-order) A poller delivers in claim order, not in commit order. This is a difference from a capture tool that reads the database log. A row can take its identifier or its `sequence` before another row and still commit after it. A reader that needs commit order must not derive it from the identifier. Order inside one aggregate or one key is the guarantee that QueueBox gives. It is enough for the outbox pattern, because one aggregate has one writer. ## The SQL Server pull claim serializes per source [Section titled “The SQL Server pull claim serializes per source”](#the-sql-server-pull-claim-serializes-per-source) On SQL Server, the pull claim takes an exclusive application lock on the source name. Every claim on one source therefore runs one at a time. The per-aggregate reservation still works under this lock. * Two handlers overlap only when a handler takes longer than the claim round trip. A short handler can look fully serialized. * Measurement gives a ceiling near 110 to 140 claims per second per source. More workers on one source do not raise that ceiling. * Scale a busy SQL Server source with more source names, not with more workers on one source. PostgreSQL does not have this limit. Its claim uses `SKIP LOCKED`, which never blocks a concurrent claim, so workers on one source divide the work. Size worker counts per database. A PostgreSQL sizing plan does not carry over to SQL Server, and the reverse is also true. ### The claim transaction on SQL Server [Section titled “The claim transaction on SQL Server”](#the-claim-transaction-on-sql-server) The SQL Server claim statement opens its own transaction and commits it. It rolls back on the lock-failure path. Keep the claim alone in that transaction: * Put no other application work in the claim transaction. * Commit the claim before any handler runs. A caller that breaks these rules leaves the claim uncommitted until its wider transaction commits. It loses the whole wider transaction on a lock failure. It also holds the per-source lock for the rest of that wider transaction. The C#, Go and TypeScript client libraries follow these rules. ### The driver timeout on SQL Server [Section titled “The driver timeout on SQL Server”](#the-driver-timeout-on-sql-server) Set the driver request or command timeout to at least 30 seconds. The application lock of the claim times out at 10 seconds and raises Msg 51000. A shorter driver timeout aborts the call before the server raises Msg 51000. The abandoned transaction then holds the per-source lock until the connection resets, and every claim on that source stalls. Note Treat Msg 51000 as transient. Back off before the retry, and never retry the call at once. The client libraries do this. # Authenticate requests > Protect an inbox source and the admin endpoint with a bearer token, an API key or an HMAC signature, and send credentials to a destination. This guide shows how to make QueueBox check the credentials of an incoming request, on an HTTP inbox source and on the admin endpoint. It also shows how QueueBox sends credentials to an HTTP destination. An inbox source without `auth` accepts every request. A request that fails the check gets `401 Unauthorized`, and QueueBox stores nothing. ## Protect an inbox source [Section titled “Protect an inbox source”](#protect-an-inbox-source) Add an `auth` block to the source. Choose one of three types. * Bearer token ```yaml sources: orders: type: http path: /orders idempotencyKeyPath: $.id eventTypePath: $.type auth: type: bearer token: ${ORDERS_WEBHOOK_TOKEN} ``` The sender puts the token in the `Authorization` header: ```bash curl -X POST http://localhost:8080/inbox/orders \ -H "Authorization: Bearer $ORDERS_WEBHOOK_TOKEN" \ -H 'Content-Type: application/json' \ -d '{"id":"order-1","type":"order.created"}' ``` The scheme name `Bearer` matches in any letter case. A header without a scheme fails. * API key ```yaml sources: partner: type: http path: /partner idempotencyKeyPath: $.id eventTypePath: $.type auth: type: api-key headerName: X-API-Key key: ${PARTNER_API_KEY} ``` The sender puts the key in the header that `headerName` names. The default is `X-API-Key`. ```bash curl -X POST http://localhost:8080/inbox/partner \ -H "X-API-Key: $PARTNER_API_KEY" \ -H 'Content-Type: application/json' \ -d '{"id":"evt-1","type":"partner.updated"}' ``` * HMAC signature ```yaml sources: github: type: http path: /github idempotencyKeyPath: $.delivery eventTypePath: $.action auth: type: hmac secret: ${GITHUB_WEBHOOK_SECRET} headerName: X-Hub-Signature-256 algorithm: HmacSHA256 signaturePrefix: "sha256=" ``` QueueBox computes the HMAC of the raw request body with `secret`. It writes the result as lower case hexadecimal after `signaturePrefix`, and compares it with the header. This matches the `X-Hub-Signature-256` header that GitHub sends. A sender computes the same value: ```bash body='{"delivery":"d-1","action":"opened"}' signature="sha256=$(printf '%s' "$body" | openssl dgst -sha256 -hmac "$GITHUB_WEBHOOK_SECRET" | sed 's/^.* //')" curl -X POST http://localhost:8080/inbox/github \ -H "X-Hub-Signature-256: $signature" \ -H 'Content-Type: application/json' \ -d "$body" ``` QueueBox compares every credential in constant time. A stored inbox row never holds the `Authorization` header, the `Proxy-Authorization` header, the `Cookie` header, or the header that `auth.headerName` names. ### HMAC settings [Section titled “HMAC settings”](#hmac-settings) | Field | Default | Meaning | | ------------------------ | -------------- | ----------------------------------------------------------------------------------------- | | `secret` | none, required | The shared secret. | | `headerName` | `X-Signature` | The header that carries the signature. | | `algorithm` | `HmacSHA256` | `HmacSHA256` or `HmacSHA512`. No other value loads. | | `signaturePrefix` | `sha256=` | The text before the hexadecimal signature. Set `""` for none. | | `timestampHeader` | none | A header that carries the send time, in milliseconds since the Unix epoch. | | `timestampTolerance` | `300000` | The largest difference, in milliseconds, between the send time and the clock of QueueBox. | | `signaturePayloadFormat` | see below | `body` or `timestamp-dot-body`. | ### Reject replayed requests [Section titled “Reject replayed requests”](#reject-replayed-requests) Without a timestamp, a captured request stays valid for ever. Set `timestampHeader` to make QueueBox reject an old request: ```yaml sources: billing: type: http path: /billing idempotencyKeyPath: $.id eventTypePath: $.type auth: type: hmac secret: ${BILLING_WEBHOOK_SECRET} headerName: X-Signature signaturePrefix: "sha256=" timestampHeader: X-Timestamp timestampTolerance: 300000 ``` When `timestampHeader` is set, `signaturePayloadFormat` defaults to `timestamp-dot-body`. The sender then signs the timestamp, a dot, and the body. A request with a fresh timestamp and an old signature fails. ```bash body='{"id":"inv-1","type":"invoice.paid"}' timestamp=$(( $(date +%s) * 1000 )) signature="sha256=$(printf '%s.%s' "$timestamp" "$body" | openssl dgst -sha256 -hmac "$BILLING_WEBHOOK_SECRET" | sed 's/^.* //')" curl -X POST http://localhost:8080/inbox/billing \ -H "X-Timestamp: $timestamp" \ -H "X-Signature: $signature" \ -H 'Content-Type: application/json' \ -d "$body" ``` Set `signaturePayloadFormat: body` only for a sender that signs the body alone. QueueBox then checks the age of the timestamp, but a replay with a new timestamp header passes. `timestamp-dot-body` without a `timestampHeader` does not load. Caution Stripe signs `timestamp.body` in seconds and sends it in one `Stripe-Signature` header with the fields `t=` and `v1=`. The QueueBox HMAC check reads one signature header and a timestamp in milliseconds, so it does not verify that header. ## Protect the admin endpoint [Section titled “Protect the admin endpoint”](#protect-the-admin-endpoint) The admin endpoint runs a JSONata expression that the caller sends. QueueBox registers it only when `admin.enabled` is `true`, and it refuses to start when `admin.auth` is absent. ```yaml admin: enabled: true auth: type: bearer token: ${ADMIN_TOKEN} maxTransformTimeoutMs: 1000 maxPayloadBytes: 65536 ``` `admin.auth` takes the same `bearer` and `api-key` blocks as a source. A request without valid credentials gets `401`. ```bash curl -X POST http://localhost:8080/admin/transform/test \ -H "Authorization: Bearer $ADMIN_TOKEN" \ -H 'Content-Type: application/json' \ -d '{"expression":"$.amount * 2","payload":{"amount":21}}' ``` The admin endpoint also accepts `hmac`. Sign the raw request body in the same way as for a source. QueueBox 0.4.0 and earlier answer `500` to every HMAC request on the admin endpoint, so use `bearer` or `api-key` there on those versions. `admin.insecure: true` allows the endpoint with no authentication. Use it on a local machine only. Set `server.managementPort` to move `/admin`, `/health` and `/metrics` to a separate port, and keep that port on an internal network. See [Security](/operations/security/). ## Authenticate to a destination [Section titled “Authenticate to a destination”](#authenticate-to-a-destination) An HTTP destination can send credentials on every request. * OAuth2 ```yaml destinations: protected-api: type: http baseUrl: https://api.example.com path: /events auth: type: oauth2 clientId: queuebox clientSecret: ${CLIENT_SECRET} tokenUrl: https://auth.example.com/oauth/token scope: api:write extraParams: audience: https://api.example.com routes: - topicPattern: "**" destination: protected-api ``` QueueBox runs the client credentials flow against `tokenUrl`, and it sends the access token as a bearer token. `scope` and `extraParams` are optional. * Basic ```yaml destinations: legacy-api: type: http baseUrl: https://legacy.example.com auth: type: basic username: queuebox password: ${LEGACY_API_PASSWORD} routes: - topicPattern: "**" destination: legacy-api ``` * Header ```yaml destinations: static-api: type: http baseUrl: https://static.example.com auth: type: header headerName: Authorization headerValue: "Bearer ${STATIC_TOKEN}" routes: - topicPattern: "**" destination: static-api ``` `headerName` defaults to `Authorization`. A row header of the same name wins over the authentication header. Do not write credential headers into outbox rows. ## Keep secrets out of the file [Section titled “Keep secrets out of the file”](#keep-secrets-out-of-the-file) Every secret field accepts an environment variable, such as `${ADMIN_TOKEN}`, or a `file:` reference. A `file:` reference reads a mounted file, for example a Kubernetes secret, once at startup: ```yaml # fragment sources: orders: auth: type: bearer token: file:/var/run/secrets/queuebox/orders-token ``` QueueBox prints a mask in place of every secret. `database.url`, the RabbitMQ destination `url` and the RabbitMQ source `connectionUrl` take no `file:` reference, because a log needs their host and port. QueueBox masks the password inside them. Set those through environment variables. # Bridge a broker > Read RabbitMQ, Kafka or NATS JetStream into the inbox, and publish outbox rows to them. This guide shows how to connect QueueBox to RabbitMQ, Kafka or NATS JetStream. A broker can feed the inbox as a source, and it can receive outbox rows as a destination. Each broker has a runnable example in the repository: `examples/rabbitmq-bridge`, `examples/kafka-bridge` and `examples/nats-bridge`. | Broker | As a source | As a destination | | -------- | ----------------------------- | ---------------------------- | | RabbitMQ | One queue per source | One exchange per destination | | Kafka | One consumer group per source | One topic per destination | | NATS | JetStream only | JetStream by default | ## Read a broker into the inbox [Section titled “Read a broker into the inbox”](#read-a-broker-into-the-inbox) A broker source stores each message as an inbox row, and then acknowledges it. A crash between the two replays the message. The unique constraint on `(source, idempotency_key)` then rejects the repeat, so the inbox holds one row. A message whose body is not JSON is stored as a `dead` row and acknowledged. No consumer can read such a body, and a message that the source never acknowledges blocks the queue or the partition. * RabbitMQ ```yaml sources: orders-queue: type: rabbitmq queueName: incoming-orders connectionUrl: amqp://guest:guest@rabbitmq:5672 idempotencyKeyPath: $.messageId aggregateIdPath: $.orderId eventTypePath: $.type topic: "orders.{{ eventType }}" prefetchCount: 10 ``` * The source acknowledges after the inbox row commits, and it requeues the message when the store fails. * The source does not declare its queue. A missing queue stops the source with an error that names it. A declared queue with a typo in its name receives nothing and looks healthy. Set `declareQueue: true` only when QueueBox must create the queue. * `prefetchCount` limits the unacknowledged messages per consumer. The default is `10`. * Kafka ```yaml sources: orders-topic: type: kafka bootstrapServers: broker-1:9092,broker-2:9092 topics: [orders, orders-retry] groupId: queuebox-orders idempotencyKeyPath: $.id aggregateIdPath: $.customerId eventTypePath: $.type autoOffsetReset: earliest maxPollRecords: 100 ``` * The source commits an offset only after the inbox row commits. It commits the unbroken run of stored records, and it seeks back to the first record that failed. * Every replica shares `groupId`, so the replicas divide the partitions between them. * `autoOffsetReset` is `earliest` or `latest`. It applies when the group has no committed offset. * `securityProtocol` is `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT` or `SASL_SSL`. A SASL protocol needs `saslMechanism`, `saslUsername` and `saslPassword`. * NATS ```yaml sources: orders-stream: type: nats servers: nats://nats:4222 stream: ORDERS durable: queuebox-orders filterSubject: orders.> idempotencyKeyPath: $.id aggregateIdPath: $.customerId eventTypePath: $.type ackWaitMs: 30000 batchSize: 100 ``` * The source reads JetStream only. Core NATS acknowledges nothing, so an inbox on it loses every message that arrives during a restart. * Create the stream before QueueBox starts. QueueBox never creates a stream, because the retention and the replication of a stream are your decisions. * `durable` names the durable consumer, which keeps its position across a restart. * The source acknowledges after the inbox row commits. It sends a negative acknowledgement when the store fails, so JetStream returns the message at once. After a crash, JetStream returns the message after `ackWaitMs`. * `servers` takes several servers, separated by commas. `username` with `password`, or `token`, sets the credentials. ### Give every message an idempotency key [Section titled “Give every message an idempotency key”](#give-every-message-an-idempotency-key) A broker source reads the idempotency key from the first of these that gives a value: | Order | RabbitMQ | Kafka | NATS | | ----- | -------------------------------- | -------------------------------- | -------------------------------- | | 1 | Header `x-idempotency-key` | Header `x-idempotency-key` | Header `x-idempotency-key` | | 2 | `idempotencyKeyPath` in the body | `idempotencyKeyPath` in the body | `idempotencyKeyPath` in the body | | 3 | AMQP `messageId` property | Record key | Header `Nats-Msg-Id` | | 4 | SHA-256 digest of the body | SHA-256 digest of the body | SHA-256 digest of the body | `idempotencyKeyPath` defaults to `$.id`. The digest deduplicates a redelivery correctly. It also merges two different events that carry the same body, and QueueBox then drops the second one. Give every message a key. ### Set the topic of a relayed message [Section titled “Set the topic of a relayed message”](#set-the-topic-of-a-relayed-message) A push source forwards each row into the outbox. The `topic` template of the source sets the outbox topic. It can read `{{ source }}` and `{{ eventType }}`, and the default for a broker source is `{{ source }}`. The event type comes from `eventTypePath` in the body first. When the body gives nothing, it comes from the header `x-event-type`. A message with no event type renders `{{ eventType }}` as an empty string, and the relay then marks it `dead`. QueueBox refuses to start when a broker source template reads `{{ eventType }}` and the source sets neither `eventTypePath` nor `eventTypeFromHeader: true`. Set `eventTypeFromHeader: true` only when every publisher sets the header. The aggregate identifier comes from `aggregateIdPath` first, then from the header `x-aggregate-id`. A Kafka source then falls back to the record key. ### Rename the attribute headers [Section titled “Rename the attribute headers”](#rename-the-attribute-headers) `attributeHeaders` renames the three headers that a broker source reads. The defaults are `x-idempotency-key`, `x-aggregate-id` and `x-event-type`. A Debezium producer, for example, sends `id`, `aggregateId` and `eventType`: ```yaml sources: outbox-events: type: kafka bootstrapServers: broker-1:9092 topics: [outbox.event.orders] groupId: queuebox-orders attributeHeaders: idempotencyKey: id aggregateId: aggregateId eventType: eventType ``` The setting changes the header names only. It does not change the order in the table above. ### Filter on headers [Section titled “Filter on headers”](#filter-on-headers) A `filter` block drops a message before QueueBox stores it. A dropped message is acknowledged, and no row is stored. ```yaml sources: orders: type: rabbitmq queueName: incoming-orders connectionUrl: amqp://guest:guest@rabbitmq:5672 idempotencyKeyPath: $.messageId filter: require: - header: x-tenant equals: acme exclude: - header: x-test exists: true ``` A message must match every `require` rule and no `exclude` rule. Each rule sets exactly one of `equals`, `in`, `matches` or `exists: true`. [Configuration](/reference/configuration/) lists the full rules. ## Publish outbox rows to a broker [Section titled “Publish outbox rows to a broker”](#publish-outbox-rows-to-a-broker) A route sends a row to a broker destination. The destination marks the row `sent` only after the broker confirms the publish. * RabbitMQ ```yaml destinations: events-exchange: type: rabbitmq url: amqp://guest:guest@rabbitmq:5672 exchange: queuebox-events exchangeType: topic deliveryMode: persistent routes: - topicPattern: "order.**" destination: events-exchange routingKeyTemplate: "{{ topic }}" ``` * The publisher declares its exchange. `exchangeType` is `topic` by default. * The routing key comes from the `routingKeyTemplate` of the route. When the route sets none, it comes from the destination field of the same name, which defaults to `{{ topic }}`. * `deliveryMode: persistent` is the default. A persistent message in a durable queue survives a broker restart. `transient` does not. * The publisher waits for one broker confirm per message, on one channel per destination. More `outbox.concurrency` raises the throughput across destinations, not inside one. * Kafka ```yaml destinations: orders-processed: type: kafka bootstrapServers: broker-1:9092,broker-2:9092 topic: orders-processed keyTemplate: "{{ key }}" timeoutMs: 30000 routes: - topicPattern: "order.**" destination: orders-processed ``` * The producer publishes with `acks=all` and idempotence. A row is `sent` only after every in-sync replica holds the record. * `keyTemplate` sets the record key. The default `{{ key }}` uses the outbox `key`, so the rows of one key land in one partition. An empty result sends no record key. * `timeoutMs` is the whole publish budget. It must be at least `2000`. * `headers` adds static record headers. The row headers travel as record headers too. * NATS ```yaml destinations: orders-processed: type: nats servers: nats://nats:4222 subject: processed.orders jetStream: true timeoutMs: 30000 routes: - topicPattern: "order.**" destination: orders-processed ``` * `jetStream: true` is the default. The publish waits for the JetStream acknowledgement, so a stream must hold the subject. * `jetStream: false` publishes on core NATS with no acknowledgement. QueueBox then marks a row `sent` without proof that anything received it. * A route `routingKeyTemplate` does not change the subject. Use the `subject` template or `subjectFrom`. ### Choose the address from the row [Section titled “Choose the address from the row”](#choose-the-address-from-the-row) The RabbitMQ `exchange`, the Kafka `topic` and the NATS `subject` can each be a template. A template can read `{{ topic }}`, `{{ key }}`, `{{ aggregateType }}`, and a payload field through `{{ payload.field }}` or `{{ data.field }}`. ```yaml destinations: domain-events: type: kafka bootstrapServers: broker-1:9092 topic: "public.{{ aggregateType }}.v1" routes: - topicPattern: "**" destination: domain-events ``` A row with `aggregate_type` set to `order` goes to the Kafka topic `public.order.v1`. `exchangeFrom`, `topicFrom` and `subjectFrom` read a row column instead, and they win over the template. The permitted columns are `aggregate_type`, `topic` and `key`, spelled with the underscore. QueueBox refuses to start when a template or a column name falls outside these sets. The error names the value and the destination. Note A column or a template that gives an empty address fails the publish. The row then follows the retry path, and it goes to `dead` at its retry ceiling. ## Next steps [Section titled “Next steps”](#next-steps) * [Write outbox rows](/how-to/write-outbox-rows/) shows how an application sets `key`, `headers` and `aggregate_type`. * [Transform payloads](/how-to/transform-payloads/) reshapes a payload on its way in or out. * [Delivery semantics](/concepts/delivery-semantics/) states what each broker path promises. # Capture changes > Turn on PostgreSQL logical capture or SQL Server change data capture, so QueueBox delivers a new outbox row without waiting for the next poll. This guide shows how to turn on change data capture for the outbox, and how to recover when capture stops. Capture reads the database log and wakes delivery when a new outbox row commits. Delivery still claims and publishes each row through SQL. [Capture](/concepts/capture/) explains the design. Capture is off by default. `outbox.capture.mode` is `polling`, and delivery polls every `outbox.pollIntervalMs`. An upgrade changes nothing until you turn capture on. | `outbox.capture.mode` | Database | Behaviour | | --------------------- | ---------- | ------------------------------------------------- | | `polling` | any | The default. Delivery polls. | | `postgres-logical` | PostgreSQL | Delivery wakes on the logical replication stream. | | `sqlserver-cdc` | SQL Server | Delivery wakes on the change data capture tables. | The connector runs inside the QueueBox process. You run no Kafka, no Kafka Connect and no Debezium Server. ## Prepare the database [Section titled “Prepare the database”](#prepare-the-database) * PostgreSQL 1. Start the server with `wal_level = logical`. 2. Give the capture account the `REPLICATION` attribute and `SELECT` on the outbox table. 3. Create the publication for the outbox table: ```sql CREATE PUBLICATION queuebox_outbox FOR TABLE outbox; ``` QueueBox never creates or drops the publication. It refuses to start capture when the publication is absent, because an automatic publication can capture the wrong tables. The connector creates the replication slot on the first start. QueueBox never drops the slot. A slot holds write-ahead log until capture reads it, so watch `pg_replication_slots`. Drop the slot by hand when you retire a capture identity. * SQL Server 1. Run the SQL Server Agent. Change data capture needs it. 2. Turn on change data capture for the database and for the outbox table: ```sql EXEC sys.sp_cdc_enable_db; EXEC sys.sp_cdc_enable_table @source_schema = N'dbo', @source_name = N'outbox', @role_name = NULL, @supports_net_changes = 0; ``` QueueBox checks both before it starts the connector. It reports a clear error instead of a connector that fails in a loop. ## Prepare the state directory [Section titled “Prepare the state directory”](#prepare-the-state-directory) Capture records its log position in `stateDirectory`. The directory must survive a restart, so mount a durable volume there. A container without a volume loses the position on every restart. | File | Content | | ------------- | -------------------------------------------------------------------------- | | `offsets.dat` | The log position that capture has handled. | | `history.dat` | The schema history. SQL Server only. | | `identity` | The identifier that ties the files to the row in `queuebox_capture_state`. | The image runs as the non-root user `queuebox`, and it ships `/var/lib/queuebox/capture` owned by that user. An empty Docker named volume mounted there takes that owner, and it needs no more work. A bind mount keeps the owner of the host directory. Change the owner of that directory to the user of the container before the first start. ## Turn capture on [Section titled “Turn capture on”](#turn-capture-on) Set the mode and point `stateDirectory` at the volume. Set `enabled: true` on one replica only. * PostgreSQL ```yaml database: type: postgresql url: jdbc:postgresql://db.internal:5432/queuebox username: queuebox password: ${DB_PASSWORD} outbox: capture: mode: postgres-logical enabled: true identity: queuebox stateDirectory: /var/lib/queuebox/capture publication: queuebox_outbox slot: queuebox_outbox reconciliationIntervalMs: 1000 ``` * SQL Server ```yaml database: type: sqlserver url: jdbc:sqlserver://db.internal:1433;databaseName=queuebox username: queuebox password: ${DB_PASSWORD} outbox: capture: mode: sqlserver-cdc enabled: true identity: queuebox stateDirectory: /var/lib/queuebox/capture reconciliationIntervalMs: 1000 connection: encrypt: true trustServerCertificate: false ``` | Setting | Default | Meaning | | -------------------------- | ----------------- | ------------------------------------------------------------------------ | | `mode` | `polling` | `polling`, `postgres-logical` or `sqlserver-cdc`. | | `enabled` | `false` | `true` on the one replica that owns capture. | | `identity` | `queuebox` | The name of the capture owner. | | `stateDirectory` | empty | The durable directory. Required when capture is on. | | `schema` | `public` or `dbo` | The schema of the outbox table. | | `publication` | `queuebox_outbox` | PostgreSQL only. The publication that you created. | | `slot` | `queuebox_outbox` | PostgreSQL only. The replication slot. | | `reconciliationIntervalMs` | `1000` | The longest wait between two delivery passes when no deadline is nearer. | The reconciliation timer keeps delivery running when capture is down or slow. A lost capture event costs time, not a message. Capture reacts to inserts and snapshot records only. It ignores updates and deletes, so the state changes that QueueBox writes create no new work. ### Use a separate connection [Section titled “Use a separate connection”](#use-a-separate-connection) Capture reads the host, the port and the database name from `database.url`. Each field under `capture.connection` replaces one of those values, for capture only. Delivery keeps `database.url`. ```yaml # fragment outbox: capture: connection: hostname: replica.db.internal port: 5433 database: queuebox username: capture_user password: ${CAPTURE_PASSWORD} ``` Use the overrides for a separate capture account, for a read replica, or for one host when `database.url` names several. QueueBox refuses a URL with more than one host unless `capture.connection.hostname` names one. `encrypt` and `trustServerCertificate` apply to SQL Server only. ## Run one owner [Section titled “Run one owner”](#run-one-owner) One process at a time owns a capture identity. The owner holds a database session lock for its whole run. A second process with the same identity fails to get the lock. It reports that the identity has an owner, and it keeps delivering through SQL. Set `enabled: false` on every other replica. To move capture to another host, stop the current owner, move or recreate the state directory, and start the new owner with the same identity. QueueBox does not choose a new owner by itself. ## Check capture health [Section titled “Check capture health”](#check-capture-health) `/health/ready` reports the component `outbox-capture`. The component is advisory: a capture fault never makes the instance unready, because delivery goes on through SQL. When the connector fails, QueueBox marks capture unhealthy and retries with a backoff from one second to thirty seconds. It wakes delivery on every attempt. A fault that needs a decision stops capture. QueueBox reports the reason and keeps delivering at the reconciliation interval. The `queuebox_capture_state` table detects these faults: * The state volume is missing or empty, but the database says that capture ran before. * The state files belong to a different instance. * The settings changed. A new slot, publication, schema, table, host or database makes the recorded position meaningless. * On PostgreSQL, the replication slot is gone while the state files remain. ## Recover capture [Section titled “Recover capture”](#recover-capture) Recovery is your decision, because an automatic answer can replay or drop log events. Use these steps after QueueBox reports that capture needs recovery: 1. Read the reason in the report. It names the fault. 2. Stop the QueueBox process that owns the capture identity. 3. If the state files are only unavailable, restore the volume and go to step 7. Capture continues from the recorded position. 4. If the state is lost, or you changed the capture settings on purpose, reset the identity: ```sql DELETE FROM queuebox_capture_state WHERE identity_name = 'queuebox'; ``` 5. On PostgreSQL, drop the slot, so the next start creates a clean one: ```sql SELECT pg_drop_replication_slot('queuebox_outbox'); ``` 6. Delete the files in `stateDirectory`. 7. Start QueueBox. After a reset, capture takes a fresh snapshot of the outbox table. A fresh snapshot only wakes delivery for rows that are still in the table. It delivers no message twice, because delivery claims each row through SQL. Change the slot, the publication, the schema or the table only together with these steps. Note `examples/cdc` in the repository runs PostgreSQL logical capture end to end. Its smoke test shows that capture wakes delivery, that the position survives a restart, and that a second owner is refused. # Consume the inbox > Choose between the push relay and a pull worker, configure a pull source, and write an idempotent handler in Go, TypeScript or C#. This guide shows the two ways an application consumes a message that an inbox source stored: the push relay and a pull worker. It then shows how to write a handler that tolerates a repeat. ## Choose push or pull [Section titled “Choose push or pull”](#choose-push-or-pull) Every source declares how its messages leave the inbox, with `consumption`. The value is `push` by default. | | `push` | `pull` | | ---------------------------------------- | ------------------------------------------------ | ----------------------------------------------- | | Who moves the message | The inbox relay inside QueueBox | Your worker | | Needs a topic, a route and a destination | Yes | No | | Where your code receives the message | At the destination, for example an HTTP endpoint | In a handler, with an open database transaction | | `state = 'processed'` means | QueueBox forwarded the row into the outbox | Your application finished the work | QueueBox stores the value on the row at receipt. A later change of the configuration does not change a message that the inbox already holds. Use push when the consumer is another service or a broker. Use pull when the consumer is your own application and its database write must commit together with the completion of the message. ## Consume with the push relay [Section titled “Consume with the push relay”](#consume-with-the-push-relay) The relay claims each stored push row, writes it into the outbox, and marks the inbox row `processed`. The outbox then routes, transforms and delivers it like any other row. Your code receives the message at the destination that the matching route names. ```yaml sources: stripe: type: http path: /stripe idempotencyKeyPath: $.id eventTypePath: $.type aggregateIdPath: $.data.object.customer topic: "stripe.{{ eventType }}" destinations: payments-service: type: http baseUrl: https://payments.internal path: /events routes: - topicPattern: "stripe.**" destination: payments-service ``` The relay builds the outbox row from the inbox row: * `topic` comes from the `topic` template of the source. The template can read `{{ source }}` and `{{ eventType }}`. An HTTP source defaults to `{{ eventType }}`. A RabbitMQ, Kafka or NATS source defaults to `{{ source }}`. * `key` is the aggregate identifier that `aggregateIdPath` reads. The rows of one aggregate therefore arrive in order. * `headers` holds the received headers, plus `x-inbox-id`, `x-source`, `x-idempotency-key` and `X-Correlation-Id`. These four replace a received header of the same name, so a sender cannot set them. * `max_attempts` is `inbox.relay.maxAttempts`, or `outbox.maxAttempts` when that is unset. A template that renders an empty topic sends the message to `dead`. This happens when the template reads `{{ eventType }}` and the message carries no event type. Set `eventTypePath`, or use `{{ source }}` in the template. The relay holds one message of an aggregate in flight at a time, across every push source together. It forwards the messages of one aggregate in the order of `created_at`. A row with no aggregate identifier takes part in no ordering. The relay has its own settings under `inbox.relay`: ```yaml inbox: relay: enabled: true pollIntervalMs: 100 batchSize: 100 claimTimeoutMs: 300000 ``` `enabled: false` stops the relay. The inbox then keeps push rows as a write-only log. Note Do not build a consumer that polls the inbox table for push rows. The relay owns those rows. Read the table to answer an operational question, for example whether a webhook arrived. ## Configure a pull source [Section titled “Configure a pull source”](#configure-a-pull-source) Set `consumption: pull` on the source. A pull source needs no `topic`, no route and no destination, because the relay never claims its rows. ```yaml sources: orders: type: http path: /orders idempotencyKeyPath: $.id aggregateIdPath: $.customerId consumption: pull ``` Rows that existed before a source changed to `pull` stay push rows. `aggregateIdPath` sets the unit of order. A pull claim holds one message in state `processing` per source and aggregate identifier. The rule holds across every worker of that source. Two sources do not block each other, so a message of aggregate `A` on `orders` can run beside a message of aggregate `A` on `payments`. ## Run a pull worker [Section titled “Run a pull worker”](#run-a-pull-worker) A client library holds the claim, the lease renewal, the completion, the retry and the dead letter. Your application writes only the handler. | Language | Package | Install | | ---------- | ------------------------------------------- | -------------------------------------------------- | | Go | `github.com/alternayte/queuebox/clients/go` | `go get github.com/alternayte/queuebox/clients/go` | | TypeScript | `@alternayte/queuebox-inbox` | `npm install @alternayte/queuebox-inbox` | | C# | `QueueBox.Inbox` | `dotnet add package QueueBox.Inbox` | Each library needs the V10 schema or later. Each works with PostgreSQL and SQL Server through the driver your application already uses. The handler receives the message and an open transaction. The library works like this: 1. It claims up to `BatchSize` messages of the source, in a transaction of its own, and commits that claim. 2. It renews the lease of each claimed message every third of the lease duration. 3. It calls the handler with the message and a new transaction. 4. When the handler returns without an error, it marks the message `processed` in that transaction and commits. 5. When the handler fails, it rolls the transaction back and asks the retry policy for a retry or a dead letter. If the completion finds no row, another worker owns the message. The library then rolls your writes back. * Go ```go worker, err := queuebox.NewInboxWorker(db, queuebox.Options{ Source: "orders", BatchSize: 10, LeaseMS: 30_000, }) if err != nil { log.Fatal(err) } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() err = worker.Run(ctx, func(ctx context.Context, message queuebox.Message, tx *sql.Tx) error { var order struct { ID string `json:"id"` Total int `json:"total"` } if err := message.UnmarshalPayload(&order); err != nil { return err } _, err := tx.ExecContext(ctx, "INSERT INTO orders (id, total) VALUES ($1, $2)", order.ID, order.Total) return err }) ``` Return an error to fail the message. Cancel the context to stop the worker. * TypeScript ```ts const worker = new InboxWorker(fromPg(pool), { source: "orders", batchSize: 10, leaseMs: 30_000 }); const stopping = new AbortController(); process.on("SIGTERM", () => stopping.abort()); await worker.run(async (message, tx, signal) => { const order = message.payload as { id: string; total: number }; await tx.query("INSERT INTO orders (id, total) VALUES ($1, $2)", [order.id, order.total]); }, stopping.signal); ``` Throw to fail the message. Abort the signal that you pass to `run` to stop the worker. * C# ```csharp var worker = new InboxWorker( InboxConnections.From(dataSource), new InboxOptions { Source = "orders", BatchSize = 10, LeaseMs = 30_000 }); await worker.RunAsync(async (message, transaction, cancellationToken) => { await using var command = transaction.CreateCommand(); command.CommandText = "INSERT INTO orders (id, total) VALUES (@id, @total)"; command .WithParameter("@id", message.Payload.GetProperty("id").GetString()) .WithParameter("@total", message.Payload.GetProperty("total").GetInt32()); await command.ExecuteNonQueryAsync(cancellationToken); }, stopping.Token); ``` Throw to fail the message. Cancel the token that you pass to `RunAsync` to stop the worker. `QueueBox.Inbox.DependencyInjection` runs a worker as an `IHostedService` through `AddQueueBoxInbox`. On a stop, the worker stops claiming at once. A running handler keeps the shutdown grace, 30 seconds by default. A handler that does not finish in that time is cancelled. The library completes nothing for it, spends no attempt on it, and lets the lease expire, so another worker takes the message. [Pull clients](/reference/pull-clients/) lists every setting and every message field of the three libraries. ### Decide what a failure does [Section titled “Decide what a failure does”](#decide-what-a-failure-does) The default policy retries with an exponential backoff and jitter, and moves the message to `dead` after five attempts. Write your own policy when some failures never succeed on a retry, for example a payload that does not parse. * Go ```go policy := queuebox.RetryPolicyFunc(func(message queuebox.Message, failure error) queuebox.FailureAction { var syntax *json.SyntaxError if errors.As(failure, &syntax) { return queuebox.DeadLetter() } if message.Attempt >= 5 { return queuebox.DeadLetter() } return queuebox.RetryAfter(time.Duration(1< { if (failure instanceof SyntaxError) { return deadLetter(); } return message.attempt >= 5 ? deadLetter() : retryAfter(2 ** message.attempt * 1000); }; const worker = new InboxWorker(fromPg(pool), { source: "orders", retryPolicy }); ``` * C# ```csharp public sealed class OrdersRetryPolicy : IInboxRetryPolicy { public InboxFailureAction Decide(InboxMessage message, Exception failure) => failure switch { JsonException => InboxFailureAction.DeadLetter(), _ when message.Attempt >= 5 => InboxFailureAction.DeadLetter(), _ => InboxFailureAction.Retry(TimeSpan.FromSeconds(Math.Pow(2, message.Attempt))), }; } var options = new InboxOptions { Source = "orders", RetryPolicy = new OrdersRetryPolicy() }; ``` A retry raises `attempt` and delays the next claim. A dead letter moves the row to `dead`, and no worker claims it again. ### Raise the concurrency [Section titled “Raise the concurrency”](#raise-the-concurrency) `MaxConcurrency` is `1` by default, so one handler runs at a time. Raise it only when the handler is safe beside a message of another aggregate. The claim never gives two workers two messages of one aggregate at the same time. ### Run on SQL Server [Section titled “Run on SQL Server”](#run-on-sql-server) Set the dialect option of the library to SQL Server. Then check two things: * Set the command or request timeout of the driver to at least 30 seconds. The claim waits up to 10 seconds for a lock and then raises Msg 51000. A shorter driver timeout aborts the call first, and the lock stays held until the connection resets. The `mssql` package for Node has a default of 15 seconds, so raise it. * The claim runs one at a time per source, behind an application lock. One source reaches about 110 to 140 claims per second, and more workers on that source do not raise it. Split a busy source into more sources. PostgreSQL has neither limit. Its claim uses `SKIP LOCKED`, so workers on one source divide the work. ### Use another language [Section titled “Use another language”](#use-another-language) `examples/pull/sql` in the repository holds the five statements that the libraries run: claim, renew, complete, retry and dead. Bind every value as a parameter. [The pull worker contract](https://github.com/alternayte/queuebox/blob/main/examples/pull/README.md) states the rules: * Run the claim alone in its own transaction, and commit it before a handler starts. * Renew every third of the lease. A renew, complete, retry or dead statement must change exactly one row. Zero rows means another worker owns the message, so stop the work. * Run the completion in the transaction of the business writes. If it changes no row, roll the whole transaction back. * On SQL Server, treat Msg 51000 as a temporary failure. Wait before the next claim. ## Write an idempotent handler [Section titled “Write an idempotent handler”](#write-an-idempotent-handler) Delivery is at least once on both paths. A lease can expire while a handler runs, and a worker can stop after its work but before the commit. The message then runs again. Write every handler so that a second run changes nothing. ### In a pull handler [Section titled “In a pull handler”](#in-a-pull-handler) 1. Write every application change through the transaction that the handler receives. The completion commits with it, so a repeat of the whole handler finds nothing done. 2. Do not commit and do not roll back. The library owns both. 3. Honour the cancellation signal. It fires when the lease is lost and when a shutdown runs out of grace. 4. Make external work, such as an HTTP call, safe to repeat. A transaction cannot roll back a call to another system. Deduplicate it on the source and the idempotency key together. The identity of a message is `(source, idempotency_key)`. Two sources can send the same event ID for two different events, so always include the source. In C#, an Entity Framework Core context opens its own connection by default. Build the context on the handler transaction with `InboxDbContextFactory.CreateOn` from `QueueBox.Inbox.DependencyInjection`. A context on a second connection commits on its own, and a later failure then repeats a write that already exists. ### At a push destination [Section titled “At a push destination”](#at-a-push-destination) A destination receives each message with an `X-Message-Id` header. Store the identifier and the effect in one transaction. A repeat then finds the identifier and changes nothing. ```sql BEGIN; INSERT INTO delivery_receipts (idempotency_key, received_at) VALUES ($1, now()) ON CONFLICT (idempotency_key) DO NOTHING; -- Zero rows inserted: a repeat. Skip the business change. -- One row inserted: new work. Apply the business change here. COMMIT; ``` Choose the key by the path of the message: * For relay traffic, use the `x-idempotency-key` header. A replay of an inbox row makes a new outbox row with a new `X-Message-Id`, but the idempotency key stays the same. * For rows that your application wrote into the outbox, use `X-Message-Id`. Answer `2xx` only after the transaction commits. QueueBox marks the row `sent` on any `2xx`, including `202`, and never sends it again. [Delivery semantics](/concepts/delivery-semantics/) explains why. # Fan out over HTTP > Deliver outbox rows to HTTP endpoints, route them on topic patterns, and send one event to several destinations. This guide shows how to deliver outbox rows to HTTP endpoints. It covers the HTTP destination, the routes that choose a destination by topic, and the way to send one event to several endpoints. `examples/http-fanout` in the repository runs the full setup. ## Declare an HTTP destination [Section titled “Declare an HTTP destination”](#declare-an-http-destination) An HTTP destination is one endpoint. QueueBox sends each row to it as a `POST` to `baseUrl` followed by `path`. ```yaml destinations: analytics: type: http baseUrl: https://analytics.internal path: /events timeoutMs: 5000 headers: X-Source: queuebox routes: - topicPattern: "analytics.**" destination: analytics ``` | Field | Default | Meaning | | ----------- | -------------- | ------------------------------------------------------------------------------------------------------------------------------- | | `baseUrl` | none, required | An absolute `http` or `https` URL with a host. QueueBox checks it at startup. | | `path` | `/` | The path after `baseUrl`. | | `timeoutMs` | `30000` | The request timeout. The connect timeout is half of it. | | `headers` | none | Static headers on every request. | | `auth` | none | OAuth2, basic or header credentials. See [Authenticate requests](/how-to/authenticate-requests/#authenticate-to-a-destination). | | `transform` | none | A JSONata expression that reshapes the payload. See [Transform payloads](/how-to/transform-payloads/). | ## What the endpoint receives [Section titled “What the endpoint receives”](#what-the-endpoint-receives) The body is the payload of the row, with `Content-Type: application/json`. QueueBox adds these headers: | Header | Value | | --------------- | ------------------------------------------------------------------------ | | `X-Message-Id` | The `id` of the outbox row. It stays the same across retries. | | `X-Topic` | The `topic` of the row. | | `X-Attempt` | The number of failed deliveries so far. It is `0` on the first delivery. | | `X-Message-Key` | The `key` of the row. It is absent when the row has no key. | The row `headers` travel as HTTP headers as well. A row header wins over a static destination header and over an authentication header of the same name. A row that the inbox relay forwards also carries `x-inbox-id`, `x-source`, `x-idempotency-key` and `X-Correlation-Id`. [Headers](/reference/headers/) lists them all. ## What the endpoint answers [Section titled “What the endpoint answers”](#what-the-endpoint-answers) * Any `2xx` completes the delivery. QueueBox marks the row `sent` and never sends it again. * Any other status fails the delivery, and so does a timeout or a connection error. QueueBox does not follow a redirect, so a `3xx` also fails. A failed delivery goes back to `pending` with a delay. The delay is `outbox.retryBaseDelayMs` times two to the power of `attempt`, plus up to 25 percent of random jitter. It never exceeds 60 seconds. When `attempt` reaches the `max_attempts` of the row, the row goes to `dead`. See [Dead letters](/operations/dead-letters/). Answer `202 Accepted` only when the message is durable at the endpoint. QueueBox treats `202` like `200` and keeps no copy to retry. An endpoint that cannot store the message yet must answer with a status outside `2xx`. Delivery is at least once. A lost response or a slow publish sends the same `X-Message-Id` again. Make the endpoint idempotent, as [Consume the inbox](/how-to/consume-the-inbox/#at-a-push-destination) shows. Set `outbox.claimTimeoutMs` above the slowest publish, or another replica takes over a row that is still in flight. ## Match topics with patterns [Section titled “Match topics with patterns”](#match-topics-with-patterns) Each route pairs a `topicPattern` with one destination. QueueBox tries the routes in configuration order, and the first route that matches wins. A row that matches no route goes to `dead`. * The pattern must match the whole topic. * `*` matches one segment. A segment holds no dot. * `**` matches any text, dots included. * Every other character matches itself. A dot, a dash and a plus have no special meaning. | Pattern | Matches | Does not match | | --------------- | ----------------------------------- | --------------------------- | | `order.*` | `order.created`, `order.paid` | `order`, `order.item.added` | | `order.**` | `order.created`, `order.item.added` | `order` | | `**` | every topic | nothing | | `order.created` | `order.created` | `order.updated` | Put the specific routes first and a catch-all `**` route last: ```yaml destinations: billing: type: http baseUrl: https://billing.internal path: /events archive: type: http baseUrl: https://archive.internal path: /events routes: - topicPattern: "invoice.*" destination: billing - topicPattern: "**" destination: archive ``` A route can also set a `transform` for its rows. It runs before the transform of the destination. ## Send one event to several endpoints [Section titled “Send one event to several endpoints”](#send-one-event-to-several-endpoints) A row goes to one destination: the destination of the first route that matches. Two routes with the same pattern do not copy a row. To reach several endpoints, write one outbox row per destination, each with its own topic. This keeps each delivery independent. Each row has its own attempt count, its own retries and its own dead-letter state. An endpoint that is down retries alone and never holds back the others. Give each destination a topic prefix: ```yaml destinations: analytics: type: http baseUrl: http://analytics:8080 path: /events timeoutMs: 5000 audit: type: http baseUrl: http://audit:8080 path: /events timeoutMs: 5000 routes: - topicPattern: "analytics.**" destination: analytics - topicPattern: "audit.**" destination: audit ``` Then write both rows in the transaction of the business write: ```sql BEGIN; INSERT INTO outbox (topic, key, payload) VALUES ('analytics.payment.succeeded', 'cust-42', '{"paymentId":"pay_1","amount":4200}'::jsonb); INSERT INTO outbox (topic, key, payload) VALUES ('audit.payment.succeeded', 'cust-42', '{"paymentId":"pay_1","amount":4200}'::jsonb); COMMIT; ``` For a message that arrives through the inbox, the source `topic` template sets the topic. One inbox message therefore reaches one destination. To fan out an inbox message, send it to one endpoint of your own that writes the rows. ## Run the example [Section titled “Run the example”](#run-the-example) `examples/http-fanout` runs QueueBox, PostgreSQL and two small HTTP receivers. Its inbox source uses the event type as the topic, so two posts reach two receivers. ```bash cd examples/http-fanout docker compose up -d --build curl -X POST http://localhost:18081/inbox/stripe -H 'Content-Type: application/json' \ -d '{"id":"e1","type":"analytics.payment.succeeded"}' curl -X POST http://localhost:18081/inbox/stripe -H 'Content-Type: application/json' \ -d '{"id":"e2","type":"audit.payment.succeeded"}' docker compose logs analytics audit docker compose down -v ``` The `analytics` log shows `e1`, and the `audit` log shows `e2`. `./smoke-test.sh` runs the same steps and checks that no receiver gets the topic of the other. ## Block private addresses [Section titled “Block private addresses”](#block-private-addresses) Set `http.blockPrivateAddresses: true` to refuse a destination whose host resolves to a loopback, link-local, site-local or unique-local address. QueueBox runs the check at startup. A host that does not resolve does not stop the start. ```yaml http: blockPrivateAddresses: true maxErrorBodyBytes: 2048 destinations: partner-api: type: http baseUrl: https://api.partner.example path: /webhooks routes: - topicPattern: "partner.**" destination: partner-api ``` `http.maxErrorBodyBytes` caps the error body that QueueBox keeps from a failed publish. QueueBox redacts secrets from that text before it reaches a log or the `last_error` column. Note Leave `blockPrivateAddresses` off when your destinations run on an internal network, as the examples above do. # Replay dead letters > Send dead or already delivered outbox messages again through the admin replay route, with filters by id, topic, destination and time. This guide replays outbox messages through the `POST /admin/replay` route. It assumes that you know how QueueBox marks a message dead. If not, read [Dead letters](/operations/dead-letters/) first. ## Choose the route or the SQL [Section titled “Choose the route or the SQL”](#choose-the-route-or-the-sql) | | `POST /admin/replay` | [SQL](/operations/dead-letters/) | | ------------------------------------ | ---------------------------------------------- | -------------------------------------- | | Selects | `sent` and `dead` outbox rows | Any row that your `WHERE` clause names | | Filters | ids, topic, topics, destination, creation time | Anything | | Resolves a destination to its topics | Yes, through the configured routes | No | | Inbox rows | No | Yes | | Needs | `admin.enabled` and admin credentials | Write access to the database | | Leaves a record | A log line with the filter and the count | Your own audit | Use the route when an operator replays by destination or by time without database access. Use the SQL for an inbox row, or for a selection that the route filters cannot express. Caution The route moves `sent` rows as well as `dead` rows. A filter by topic, destination or time sends every delivered message in that range again. To replay dead messages only, select their ids first and pass `ids`. ## Before you start [Section titled “Before you start”](#before-you-start) * Correct the cause of the failure. A replay against a destination that is still broken produces new dead messages. * Enable the admin routes with authentication. They are off by default. ```yaml # fragment admin: enabled: true auth: type: bearer token: file:/run/secrets/queuebox-admin-token ``` The admin routes are on the management port when `server.managementPort` is set. Never publish `/admin` through the ingress. See [Security](/operations/security/#the-admin-routes). ## Replay dead messages by id [Section titled “Replay dead messages by id”](#replay-dead-messages-by-id) 1. List the dead messages that you want to replay. ```sql SELECT id, topic, key, attempt, updated_at, last_error FROM outbox WHERE state = 'dead' AND topic = 'order.created' ORDER BY updated_at DESC; ``` 2. Send their ids to the route. ```bash curl -X POST http://localhost:9090/admin/replay \ -H "Authorization: Bearer $ADMIN_TOKEN" \ -H "Content-Type: application/json" \ -d '{"ids": ["3f2b1c1e-8d4a-4c1b-9a53-2d0f6f1e7a10", "9b7e5d2c-1a3f-4e8b-b6c4-0f1e2d3c4b5a"]}' ``` 3. Read the count in the answer. ```json {"moved": 2} ``` 4. Watch the rows reach `sent`. ```sql SELECT id, state, attempt, last_error FROM outbox WHERE id IN ('3f2b1c1e-8d4a-4c1b-9a53-2d0f6f1e7a10', '9b7e5d2c-1a3f-4e8b-b6c4-0f1e2d3c4b5a'); ``` ## Replay everything for one destination in a time window [Section titled “Replay everything for one destination in a time window”](#replay-everything-for-one-destination-in-a-time-window) Use this form after a destination lost data and needs every message of a period again. ```bash curl -X POST http://localhost:9090/admin/replay \ -H "Authorization: Bearer $ADMIN_TOKEN" \ -H "Content-Type: application/json" \ -d '{ "destination": "billing-service", "createdAfter": "2026-09-24T08:00:00Z", "createdBefore": "2026-09-24T12:00:00Z" }' ``` QueueBox resolves the destination to the topics that route to it. It uses the configured routes and their first-match rule, exactly as the poller does. It then moves the `sent` and `dead` rows of those topics in the window. ## The filter [Section titled “The filter”](#the-filter) Every field is optional, but the request needs at least one. The route combines the fields with AND. | Field | Type | Selects | | --------------- | --------------------- | ------------------------------------------------- | | `ids` | Array of UUID strings | Rows with these `id` values. | | `topic` | String | Rows with this exact topic. | | `topics` | Array of strings | Rows with any of these topics. | | `destination` | String | Rows whose topic routes to this destination name. | | `createdAfter` | ISO 8601 instant | Rows with `created_at` strictly after this time. | | `createdBefore` | ISO 8601 instant | Rows with `created_at` strictly before this time. | * The route always adds `state IN ('sent', 'dead')`. No filter can move a `pending` or `processing` row. * `topics` together with `destination` narrows the destination’s topics to the ones you list. It never adds a topic. When the two sets do not overlap, the route moves nothing. * A destination that no topic routes to moves nothing. * The route ignores an unknown field. Check the spelling of each field name, because a misspelt field filters nothing. ## What a replay changes [Section titled “What a replay changes”](#what-a-replay-changes) For each selected row, the replay sets: * `state` to `pending`, * `attempt` to 0, * `scheduled_at` to now, * `last_error`, `claimed_at`, `claim_token` and `lease_expires_at` to null. The row keeps its `id`, its `sequence`, its payload and its headers. The poller claims it on its next cycle and delivers it to the destination that its topic routes to now. * **Order.** A replayed row with a `key` becomes the head of its key again. It holds back later pending rows of that key until it is sent or dead. It reaches the destination after the rows that passed it. See [Ordering](/concepts/ordering/#order-and-the-key). * **Identity.** A replayed row keeps its `X-Message-Id`, because it is the same outbox row. A receiver that deduplicates on `X-Message-Id` discards the replay of a row that it already stored. Clear the receipt at the receiver first if the receiver must apply the message again. ## The answers [Section titled “The answers”](#the-answers) | Status | Body | Meaning | | ------ | -------------------------------------------------- | ---------------------------------------------------------------------------------------------- | | 200 | `{"moved": n}` | The replay moved `n` rows. `0` means that nothing matched. | | 400 | `{"error": "A replay needs at least one filter."}` | The request has no filter. | | 400 | `{"error": "Invalid request: ..."}` | The body is not valid JSON, or a field has the wrong type. | | 401 | `{"error": "..."}` | The credentials are missing or wrong. | | 413 | `{"error": "Request body exceeds ... bytes"}` | The body is larger than `admin.maxPayloadBytes` (default 65536). | | 404 | None | `admin.enabled` is false, or the request went to the data port while a management port is set. | QueueBox logs every replay with its filter and the count. An operator who loses the answer can find the replay in the log. ## Replay a dead inbox message [Section titled “Replay a dead inbox message”](#replay-a-dead-inbox-message) The route replays outbox rows only. To requeue a dead inbox row, use the SQL in [Dead letters](/operations/dead-letters/#requeue-a-dead-inbox-message). # Transform payloads > Reshape a payload with JSONata at ingestion, on a route or at a destination, choose what a failed transform does, and build RabbitMQ routing keys from the row. This guide shows how to reshape a payload with a [JSONata](https://jsonata.org/) expression, and how to build a RabbitMQ routing key from the row. [Transforms](/reference/transforms/) lists every context variable and every setting. ## Choose where the transform runs [Section titled “Choose where the transform runs”](#choose-where-the-transform-runs) A transform can run at three places. Each one sees the output of the one before it. | Place | Runs | Typical use | | ------------------------------- | ------------------------------------------- | ------------------------------------------------- | | `sources..transform` | Once, when the inbox stores the message | Normalise a webhook before anything else reads it | | `routes[n].transform` | Before delivery, for the rows of that route | Shape one kind of event | | `destinations..transform` | Before delivery, after the route transform | Fit the format that one endpoint expects | A source transform changes the stored payload. A route or destination transform changes only what the destination receives. The outbox row keeps its original payload, so a retry transforms it again. ## Reshape a payload at a destination [Section titled “Reshape a payload at a destination”](#reshape-a-payload-at-a-destination) This destination wraps every payload in an envelope: ```yaml destinations: partner-api: type: http baseUrl: https://api.partner.example path: /events transform: expression: '{ "id": $messageId, "type": $topic, "data": $ }' timeoutMs: 100 onError: Fail routes: - topicPattern: "order.*" destination: partner-api ``` A row with topic `order.created` and payload `{"orderId":"1001"}` arrives as: ```json { "id": "6b1f0c1e-9a7d-4f0e-8a1c-2f3d4e5f6a7b", "type": "order.created", "data": { "orderId": "1001" } } ``` `$` is the payload. `$messageId`, `$topic`, `$attempt` and `$timestamp` are context variables. An inbox transform also reads `$source` and `$headers`. ## Shape one kind of event on a route [Section titled “Shape one kind of event on a route”](#shape-one-kind-of-event-on-a-route) A route transform runs only for the rows that the route matches: ```yaml destinations: billing: type: http baseUrl: https://billing.internal path: /orders routes: - topicPattern: "order.*" destination: billing transform: expression: | { "orderId": id, "customer": customer.name, "total": items.(price * quantity) ~> $sum() } timeoutMs: 150 onError: Fail ``` The payload `{"id":"1001","customer":{"name":"Ada"},"items":[{"price":10,"quantity":2},{"price":5,"quantity":3}]}` becomes `{"orderId":"1001","customer":"Ada","total":35}`. ## Normalise a webhook at the source [Section titled “Normalise a webhook at the source”](#normalise-a-webhook-at-the-source) A source transform runs once, before QueueBox stores the row. It can read the received headers through `$headers`: ```yaml sources: github: type: http path: /github idempotencyKeyPath: $.delivery eventTypePath: $.action transform: expression: '{ "action": action, "repository": repository.full_name, "event": $headers."x-github-event" }' onError: Skip ``` * Quote a header name that holds a hyphen: `$headers."x-github-event"`. * A header name matches exactly, so use the letter case that the stored row shows. * An HTTP source stores no `Authorization`, `Proxy-Authorization` or `Cookie` header, and no header that the source authentication reads. * QueueBox reads the idempotency key and the event type from the original body, before the transform runs. ## More expressions [Section titled “More expressions”](#more-expressions) Remove secret fields: ```text $ ~> |$|{}, ['password', 'secret', 'token']| ``` Choose an action from a field: ```text status = 'paid' ? { "action": "fulfill", "orderId": id } : { "action": "remind", "orderId": id } ``` Add the time of the delivery: ```text $merge([$, { "deliveredAt": $timestamp }]) ``` ## Choose what a failed transform does [Section titled “Choose what a failed transform does”](#choose-what-a-failed-transform-does) `onError` decides what happens when the expression fails or runs past `timeoutMs`. Write the value with the exact case shown. `fail` in lower case does not load. | `onError` | At a route or a destination | At a source | | --------- | ------------------------------------------------- | ------------------------------------- | | `Fail` | The default. The row goes back to the retry path. | QueueBox rejects the message. | | `Skip` | The original payload goes to the destination. | QueueBox stores the original payload. | | `Dead` | The row goes to `dead` at once. | Same as `Fail`. | A rejected message at a source follows the source type: * An HTTP source answers `422` and stores no row. The sender still holds the message. * A RabbitMQ, Kafka or NATS source stores the row with the original payload, marks it `dead`, and acknowledges the message. The message is never lost. `Skip` on a route transform delivers the payload as it was before that transform, and the destination transform does not run. `Skip` on a destination transform delivers the output of the route transform. `timeoutMs` defaults to `100`. `maxDepth` defaults to `100` and bounds the recursion of one expression. ## Test an expression [Section titled “Test an expression”](#test-an-expression) The admin endpoint runs an expression against a sample payload, so you can check it before you deploy it. Turn the endpoint on with authentication, as [Authenticate requests](/how-to/authenticate-requests/#protect-the-admin-endpoint) shows. Then send the expression: ```bash curl -X POST http://localhost:8080/admin/transform/test \ -H "Authorization: Bearer the-admin-token" \ -H "Content-Type: application/json" \ -d '{ "expression": "{ \"total\": items.(price * qty) ~> $sum() }", "payload": {"items": [{"price": 10, "qty": 2}, {"price": 5, "qty": 3}]}, "mockTopic": "order.created" }' ``` ## Build a RabbitMQ routing key [Section titled “Build a RabbitMQ routing key”](#build-a-rabbitmq-routing-key) A RabbitMQ destination publishes each row with a routing key. A route sets it with `routingKeyTemplate`: ```yaml destinations: events-exchange: type: rabbitmq url: amqp://guest:guest@rabbitmq:5672 exchange: events routes: - topicPattern: "order.*" destination: events-exchange routingKeyTemplate: "{{ payload.region }}.{{ payload.priority }}.{{ topic }}" routingKeyMissingFieldDefault: "default" ``` A row with topic `order.created` and payload `{"region":"eu","priority":"high"}` gets the routing key `eu.high.order.created`. Without a `priority` field, it gets `eu.default.order.created`. A template can read these placeholders: | Placeholder | Value | | --------------------- | --------------------------------------------------------------------------- | | `{{ topic }}` | The `topic` of the row | | `{{ key }}` | The `key` of the row | | `{{ aggregateType }}` | The `aggregate_type` of the row | | `{{ payload.field }}` | A field of the payload. Nested fields work: `{{ payload.customer.region }}` | | `{{ data.field }}` | The same as `{{ payload.field }}` | A missing field and any other placeholder render as `routingKeyMissingFieldDefault`. The default of that setting is an empty string. When the route sets no `routingKeyTemplate`, QueueBox renders the `routingKeyTemplate` of the destination instead. Its default is `{{ topic }}`. Note A route `routingKeyTemplate` sets the RabbitMQ routing key only. A Kafka destination sets its record key with `keyTemplate`. A NATS destination sets its subject with `subject` or `subjectFrom`. See [Bridge a broker](/how-to/bridge-a-broker/#choose-the-address-from-the-row). # Upgrade QueueBox > Move a deployment to a new QueueBox image, apply the new migrations, and handle the releases that need every old worker stopped first. This guide shows how to move a running deployment to a new QueueBox version. It covers the image tag, the database migrations, and the two migrations that need every worker of the old version stopped first. ## Read the changelog first [Section titled “Read the changelog first”](#read-the-changelog-first) Read every entry of `CHANGELOG.md` between your version and the target version. A `Breaking` section names each change that needs action, and it states the step. The entry also names the migrations that the release ships. QueueBox follows Semantic Versioning. Below 1.0.0, a MINOR release can carry a breaking change. The changelog then lists it under `Breaking`. ## Pin the image [Section titled “Pin the image”](#pin-the-image) QueueBox ships as a container image for `linux/amd64` and `linux/arm64`. The registry carries three tags for each release: | Tag | Meaning | | ------------------------------------ | --------------------------------------- | | `ghcr.io/alternayte/queuebox:0.3.2` | The exact release. | | `ghcr.io/alternayte/queuebox:0.3` | The newest patch of that MINOR release. | | `ghcr.io/alternayte/queuebox:latest` | The newest release. | Pin the exact version in production. A published tag never moves, so a pinned deployment gets the same image on every pull. ## Upgrade with the bundled migrations [Section titled “Upgrade with the bundled migrations”](#upgrade-with-the-bundled-migrations) With `database.migrate: true`, which is the default, QueueBox applies each new migration at startup through Flyway. 1. Check that the target release needs no stop of the old workers. See [Migrations that need a full stop](#migrations-that-need-a-full-stop). 2. Change the image tag to the target version. 3. Roll the replicas to the new image. The first new replica applies the migrations. 4. Check each new replica. `GET /health/ready` answers `200`, and the `queuebox` metric carries the new version. ```bash curl -s http://localhost:8080/health/ready curl -s http://localhost:8080/metrics | grep '^queuebox{' ``` The Prometheus exporter drops the `_info` suffix, so the metric `queuebox_info` appears as `queuebox` in a scrape. Its `version` label names the running release. ## Upgrade with `database.migrate: false` [Section titled “Upgrade with database.migrate: false”](#upgrade-with-databasemigrate-false) A deployment that applies its own schema must apply each new migration before the new version starts. 1. List the migration files that the target version adds. The changelog names them, and so does [Use custom tables](/how-to/use-custom-tables/#apply-the-default-schema-by-hand). 2. Apply each new file in version order, with a privileged user. 3. For a custom schema, apply the same change to your own tables with your own names. Add the new column to the column mapping when you renamed it. 4. Roll the replicas to the new image. A new column is nullable or has a default, so the replicas of the previous version keep running against the new schema. Two migrations are exceptions, and the next section covers them. ## Migrations that need a full stop [Section titled “Migrations that need a full stop”](#migrations-that-need-a-full-stop) A migration that changes how a worker claims a row needs a full stop. The old claim and the new claim must never run at the same time. Upgrade in this order: 1. Stop every QueueBox replica of the old version. 2. Apply the migration. 3. Start the replicas of the new version. ### V6: the claim token and the lease [Section titled “V6: the claim token and the lease”](#v6-the-claim-token-and-the-lease) `V6__add_consumption_and_leases.sql` adds `claim_token` and `lease_expires_at` to both tables. It also adds `consumption`, `scheduled_at`, `attempt` and `last_error` to the inbox. An old worker fences a claim on a timestamp. A new worker fences it on a token and a lease. An old worker can therefore complete a row that a new worker owns. Stop every old worker, apply `V6__add_consumption_and_leases.sql` and then `V7__capture_state.sql`, and start the new workers. The existing inbox rows become `push` rows, which keeps their behaviour. A custom schema must add the new columns and map them. ### V11: the key order [Section titled “V11: the key order”](#v11-the-key-order) `V11__add_outbox_sequence.sql` adds the outbox `sequence` column. The new claim delivers the rows of one key in `sequence` order, one row at a time. The old claim ignores the key, so it can deliver a later row of a key before an earlier one. Stop every old replica, and apply `V11__add_outbox_sequence.sql`. The migration numbers the existing rows in `created_at` order. Then start the new replicas. V11 changes the outbox claim only, so pull workers keep running. A custom outbox table must add a `BIGINT` column that the database fills on insert, and map it as `database.columnMapping.outbox.sequence`. QueueBox stops at startup without it, and it prints the `ALTER TABLE` statement. ## Other migrations that need care [Section titled “Other migrations that need care”](#other-migrations-that-need-care) * **V8, the pull claim indexes.** `V8__add_pull_claim_indexes.sql` uses a plain `CREATE INDEX`. On both databases, the build blocks inserts into the inbox until it ends. Plan a maintenance window for a populated inbox. Or build the two indexes online first: `CREATE INDEX CONCURRENTLY` on PostgreSQL, outside a transaction, or `CREATE INDEX ... WITH (ONLINE = ON)` on SQL Server Enterprise Edition and Azure SQL. On SQL Server, run `SET QUOTED_IDENTIFIER ON` first, because the indexes are filtered. * **V10, the inbox headers.** `V10__add_inbox_headers.sql` adds the inbox `headers` column. A custom inbox table must add it by hand. QueueBox stops at startup without it, and it prints the `ALTER TABLE` statement. The pull client libraries from 0.3.0 need this column. ## Compatibility policy [Section titled “Compatibility policy”](#compatibility-policy) QueueBox keeps these promises for a MINOR release. For the configuration: * A MINOR release can add an optional field. The default keeps the previous behaviour. * A MINOR release can add a value to a field that takes a fixed set of values. * A MINOR release does not remove, rename or change the meaning of a field. * A field that QueueBox plans to remove is first marked deprecated in the changelog and in the configuration reference. The removal comes in the next MAJOR release. QueueBox prints no deprecation warning at startup. For the database schema: * A MINOR release can add a table, a column or an index. It does not drop a column or change the type of a column. * Every new column is nullable or has a default. * An instance of the previous MINOR release runs against the schema of the current MINOR release, except across V6 and V11. * A destructive change comes only in a MAJOR release. The changelog states the downtime that it needs. ## Upgrade the pull clients [Section titled “Upgrade the pull clients”](#upgrade-the-pull-clients) The client libraries release on their own tags, `csharp-v*`, `typescript-v*` and `clients/go/v*`. A client release does not need a QueueBox release. Each client README states the oldest schema that it needs. Upgrade QueueBox and apply the schema before you move a client to a version that needs a newer schema. Note Do not roll back by starting the old image against a schema that a newer version migrated past V11. The old claim ignores the key order. Correct the defect in a new patch release instead. # Use QueueBox with coding agents > Give a coding agent the QueueBox docs as Markdown and install the QueueBox skill in Claude Code, Cursor or another agent. This guide shows you how to set up a coding agent, such as Claude Code or Cursor, to write application code for QueueBox. The agent gets two sources of facts: the docs as plain Markdown, and the QueueBox skill. | Part | What the agent gets | | ---------------------------- | --------------------------------------------------------------------------------------------- | | llms.txt and the `.md` pages | The full docs as plain Markdown. | | The QueueBox skill | A short file of rules for outbox rows, `queuebox.yml`, pull clients and idempotent consumers. | ## Give the agent the docs [Section titled “Give the agent the docs”](#give-the-agent-the-docs) The docs site publishes its pages as plain text for language models: | URL | Content | | ----------------------------------------------------------------- | ---------------------------------------------------------------- | | [/llms.txt](https://queuebox-docs.pages.dev/llms.txt) | The index. It lists each page with the URL of its Markdown copy. | | [/llms-full.txt](https://queuebox-docs.pages.dev/llms-full.txt) | All pages in one Markdown file. | | [/llms-small.txt](https://queuebox-docs.pages.dev/llms-small.txt) | All pages in a compact form, for a small context window. | Each page also has a Markdown copy. Add `.md` to the path of the page. For example, [/reference/configuration.md](https://queuebox-docs.pages.dev/reference/configuration.md) is the Markdown copy of [the configuration reference](/reference/configuration/). The **Copy page** menu at the top of each page has four entries: | Entry | Effect | | -------------------- | ---------------------------------------------------------------------------------------- | | **Copy as Markdown** | Puts the Markdown of the page on the clipboard. Paste it into a chat or an agent prompt. | | **View as Markdown** | Opens the `.md` copy of the page. | | **Open in Claude** | Opens a Claude chat that reads the page. | | **Open in ChatGPT** | Opens a ChatGPT chat that reads the page. | ## Install the skill [Section titled “Install the skill”](#install-the-skill) The skill is one file: [`skills/queuebox/SKILL.md`](https://github.com/alternayte/queuebox/blob/main/skills/queuebox/SKILL.md). Its `description` tells the agent when to load it: when code writes outbox rows, consumes the inbox, receives a delivery, or changes `queuebox.yml`. The skill links to `llms-full.txt` for the detail. * Claude Code plugin The QueueBox repository is a Claude Code plugin marketplace with one plugin, `queuebox`. Run these two commands in a Claude Code session: ```text /plugin marketplace add alternayte/queuebox /plugin install queuebox@queuebox ``` The plugin ships the skill. To get a newer version of the skill, run `claude plugin marketplace update queuebox` in a shell. * npx skills The `skills` CLI reads the `skills/` directory of the repository and installs the skill for the agents that it detects: ```sh npx skills add alternayte/queuebox ``` Add `--list` to see the skills in the repository without an install. Run `npx skills update` to get a newer version. * Manual copy Copy the file into the skills directory of your repository. Claude Code finds a skill in `.claude/skills//SKILL.md`: ```sh mkdir -p .claude/skills/queuebox curl -fsSL https://raw.githubusercontent.com/alternayte/queuebox/main/skills/queuebox/SKILL.md \ -o .claude/skills/queuebox/SKILL.md ``` Commit the file, so that each person on the team gets the same rules. Copy the file again to get a newer version. * Cursor Cursor reads rules from `.cursor/rules/`. Copy the skill into a rule file: ```sh mkdir -p .cursor/rules curl -fsSL https://raw.githubusercontent.com/alternayte/queuebox/main/skills/queuebox/SKILL.md \ -o .cursor/rules/queuebox.mdc ``` Cursor reads the `description` in the front matter of the file. It attaches the rule when a request matches the description. * Other agents Many agents read `AGENTS.md` in the root of the repository. Add a line to it that points to the skill: ```md For code that uses QueueBox, read skills/queuebox/SKILL.md first. ``` Put the file at that path with one of the other methods. For an agent without `AGENTS.md` support, add the content of `SKILL.md` to its instructions file. ## What the skill covers [Section titled “What the skill covers”](#what-the-skill-covers) The skill holds the rules that an agent needs most often. The docs hold the full detail. | Section | Content | | -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | Write an outbox row | The required columns `topic` and `payload`, the optional columns, the order per `key`, `headers`, `scheduled_at` and `max_attempts`. It has an insert for Postgres and for SQL Server, with `[key]` in brackets on SQL Server. | | queuebox.yml | `database`, `sources`, `destinations`, `routes` and `transforms`, with a complete example that the QueueBox loader accepts. | | Consume the inbox | Push through the relay or pull with a client. It names the Go, TypeScript and C# packages, their core calls and the handler rules. | | Idempotent consumers | Deduplication on `X-Message-Id` for direct outbox traffic and on `x-idempotency-key` for relay traffic. | | Do not | The mistakes to avoid. For example, a write to a column that QueueBox owns, an insert outside the business transaction, or a dependency on order across keys. | ## Next steps [Section titled “Next steps”](#next-steps) [Write outbox rows](/how-to/write-outbox-rows/)Insert a message in the transaction of the business write. [Consume the inbox](/how-to/consume-the-inbox/)Read inbox messages with a pull client in Go, TypeScript or C#. [Configuration](/reference/configuration/)Every key of queuebox.yml and its environment variable. # Use custom tables > Point QueueBox at tables and columns with your own names, turn off the bundled migrations, and apply the schema by hand. This guide shows how to run QueueBox against tables that you create and name yourself. It also shows how to apply the default schema by hand when the QueueBox database user has no rights to change the schema. By default, QueueBox creates the `outbox` and `inbox` tables itself. At startup, Flyway applies the bundled migrations, because `database.migrate` is `true`. ## Choose the case [Section titled “Choose the case”](#choose-the-case) Set `database.migrate: false` and apply the schema yourself in two cases: 1. The QueueBox database user has no rights to create or change tables. Apply the bundled migration files with a privileged user. See [Apply the default schema by hand](#apply-the-default-schema-by-hand). 2. The configuration renames a table or a column. The bundled files use the default names, so they cannot create your schema. See [Rename tables and columns](#rename-tables-and-columns). QueueBox refuses to start when the configuration renames a table or a column and `database.migrate` is still `true`. The error names each renamed setting. ## Apply the default schema by hand [Section titled “Apply the default schema by hand”](#apply-the-default-schema-by-hand) Apply every migration file of your database, in version order, with a privileged user. The files are in the repository at the tag of the QueueBox version that you run. * PostgreSQL ```bash for version in 1 2 3 4 5 6 7 8 9 10 11; do psql "$ADMIN_DATABASE_URL" -v ON_ERROR_STOP=1 \ -f postgres/src/main/resources/db/postgresql/V${version}__*.sql done ``` * SQL Server ```bash for version in 1 2 3 4 5 6 7 8 9 10 11; do sqlcmd -S db.internal -d queuebox -U admin -b -I \ -i sqlserver/src/main/resources/db/sqlserver/V${version}__*.sql done ``` `-I` sets `QUOTED_IDENTIFIER ON`. `V8__add_pull_claim_indexes.sql` and `V11__add_outbox_sequence.sql` create filtered indexes, and SQL Server needs that setting for them. These are the files of the current release: | File | Change | | ------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------ | | `V1__create_outbox.sql` | Create the outbox table and its indexes. | | `V2__create_inbox.sql` | Create the inbox table, its unique constraint on `(source, idempotency_key)`, and its indexes. | | `V3__add_claimed_at.sql` | Add `claimed_at` to both tables. | | `V4__add_last_error.sql` | Add `last_error` to the outbox. | | `V5__add_correlation_id.sql` | Add `correlation_id` to the inbox. | | `V6__add_consumption_and_leases.sql` | Add `claim_token` and `lease_expires_at` to both tables, and `consumption`, `scheduled_at`, `attempt` and `last_error` to the inbox. | | `V7__capture_state.sql` | Create `queuebox_capture_state`. | | `V8__add_pull_claim_indexes.sql` | Add the two indexes of the pull claim. | | `V9__add_aggregate_type.sql` | Add `aggregate_type` to the outbox. | | `V10__add_inbox_headers.sql` | Add `headers` to the inbox. | | `V11__add_outbox_sequence.sql` | Add `sequence` to the outbox, number the existing rows, and add the index on `key` and `sequence`. | Apply every file. A missing file does not stop the start. The first insert that names the missing column fails instead. Then turn the migrations off: ```yaml database: url: jdbc:postgresql://db.internal:5432/queuebox username: queuebox_app password: ${DB_PASSWORD} migrate: false ``` A new QueueBox version can add a migration. Apply it by hand before you start that version. [Upgrade QueueBox](/how-to/upgrade/) lists the steps. ## Rename tables and columns [Section titled “Rename tables and columns”](#rename-tables-and-columns) `database.outboxTableName` and `database.inboxTableName` rename the tables. `database.columnMapping.outbox` and `database.columnMapping.inbox` rename the columns. A key that you omit keeps its default name. ```yaml database: url: jdbc:postgresql://db.internal:5432/app username: app password: ${DB_PASSWORD} migrate: false outboxTableName: qb_outbox inboxTableName: qb_inbox columnMapping: outbox: payload: body inbox: idempotencyKey: dedup_key ``` ### The column keys [Section titled “The column keys”](#the-column-keys) Each list below is complete. Every column must exist in your table, because QueueBox reads or writes each one. | Outbox key | Default column | | ---------------- | ------------------ | | `id` | `id` | | `topic` | `topic` | | `key` | `key` | | `aggregateType` | `aggregate_type` | | `payload` | `payload` | | `headers` | `headers` | | `state` | `state` | | `attempt` | `attempt` | | `maxAttempts` | `max_attempts` | | `scheduledAt` | `scheduled_at` | | `createdAt` | `created_at` | | `updatedAt` | `updated_at` | | `claimedAt` | `claimed_at` | | `claimToken` | `claim_token` | | `leaseExpiresAt` | `lease_expires_at` | | `lastError` | `last_error` | | `sequence` | `sequence` | | Inbox key | Default column | | ---------------- | ------------------ | | `id` | `id` | | `source` | `source` | | `idempotencyKey` | `idempotency_key` | | `aggregateId` | `aggregate_id` | | `eventType` | `event_type` | | `payload` | `payload` | | `headers` | `headers` | | `state` | `state` | | `consumption` | `consumption` | | `attempt` | `attempt` | | `lastError` | `last_error` | | `scheduledAt` | `scheduled_at` | | `createdAt` | `created_at` | | `processedAt` | `processed_at` | | `claimedAt` | `claimed_at` | | `claimToken` | `claim_token` | | `leaseExpiresAt` | `lease_expires_at` | | `correlationId` | `correlation_id` | ### Two columns that QueueBox checks at startup [Section titled “Two columns that QueueBox checks at startup”](#two-columns-that-queuebox-checks-at-startup) QueueBox reads the columns of both tables at startup and stops when one of these two is absent. The error names the table and the column, and it prints the `ALTER TABLE` statement for your database. * The inbox `headers` column. It holds the received headers as a JSON object. It is `NOT NULL` with the default `'{}'`. * The outbox `sequence` column. It is a `BIGINT` that the database fills on insert. The claim orders the rows of one key by it. - PostgreSQL ```sql ALTER TABLE "qb_inbox" ADD COLUMN "headers" JSONB NOT NULL DEFAULT '{}'; ALTER TABLE "qb_outbox" ADD COLUMN "sequence" BIGINT GENERATED BY DEFAULT AS IDENTITY; CREATE INDEX ON "qb_outbox" ("key", "sequence") WHERE "state" IN ('pending', 'processing'); ``` - SQL Server ```sql ALTER TABLE [qb_inbox] ADD [headers] NVARCHAR(MAX) NOT NULL DEFAULT '{}'; ALTER TABLE [qb_outbox] ADD [sequence] BIGINT IDENTITY(1,1) NOT NULL; CREATE INDEX ix_qb_outbox_key_sequence ON [qb_outbox] ([key], [sequence]); ``` QueueBox checks no other column at startup. A missing column fails the first statement that names it. ### Create the tables [Section titled “Create the tables”](#create-the-tables) The statements below create both tables with the columns, the constraints and the indexes of the current default schema. They match the configuration above: the tables are `qb_outbox` and `qb_inbox`, the outbox payload column is `body`, and the inbox idempotency key column is `dedup_key`. Change the names to match your own configuration. * PostgreSQL ```sql CREATE TABLE qb_outbox ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), topic VARCHAR(255) NOT NULL, key VARCHAR(255), aggregate_type VARCHAR(255), body JSONB NOT NULL, headers JSONB NOT NULL DEFAULT '{}', state VARCHAR(50) NOT NULL DEFAULT 'pending', attempt INTEGER NOT NULL DEFAULT 0, max_attempts INTEGER NOT NULL DEFAULT 5, scheduled_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, claimed_at TIMESTAMPTZ, claim_token UUID, lease_expires_at TIMESTAMPTZ, last_error TEXT, sequence BIGINT GENERATED BY DEFAULT AS IDENTITY ); CREATE INDEX qb_outbox_pending_scheduled ON qb_outbox (state, scheduled_at) WHERE state = 'pending'; CREATE INDEX qb_outbox_processing_claimed ON qb_outbox (claimed_at) WHERE state = 'processing'; CREATE INDEX qb_outbox_key_sequence ON qb_outbox (key, sequence) WHERE state IN ('pending', 'processing'); CREATE INDEX qb_outbox_topic ON qb_outbox (topic); CREATE TABLE qb_inbox ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), source VARCHAR(255) NOT NULL, dedup_key VARCHAR(255) NOT NULL, aggregate_id VARCHAR(255), event_type VARCHAR(255), payload JSONB NOT NULL, headers JSONB NOT NULL DEFAULT '{}', state VARCHAR(50) NOT NULL DEFAULT 'pending', consumption VARCHAR(4) NOT NULL DEFAULT 'push' CHECK (consumption IN ('push', 'pull')), attempt INT NOT NULL DEFAULT 0, last_error TEXT, scheduled_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, processed_at TIMESTAMPTZ, claimed_at TIMESTAMPTZ, claim_token UUID, lease_expires_at TIMESTAMPTZ, correlation_id VARCHAR(128), CONSTRAINT qb_inbox_source_idempotency UNIQUE (source, dedup_key) ); CREATE INDEX qb_inbox_pending ON qb_inbox (state) WHERE state = 'pending'; CREATE INDEX qb_inbox_state_created ON qb_inbox (state, created_at); CREATE INDEX qb_inbox_aggregate_state ON qb_inbox (aggregate_id, state); CREATE INDEX qb_inbox_processing_claimed ON qb_inbox (claimed_at) WHERE state = 'processing'; CREATE INDEX qb_inbox_consumption_pending ON qb_inbox (consumption, state, scheduled_at); CREATE INDEX qb_inbox_pull_pending ON qb_inbox (source, scheduled_at, created_at, id) WHERE consumption = 'pull' AND state = 'pending'; CREATE INDEX qb_inbox_pull_busy ON qb_inbox (source, aggregate_id, lease_expires_at) WHERE consumption = 'pull' AND state = 'processing'; ``` * SQL Server ```sql SET QUOTED_IDENTIFIER ON; CREATE TABLE qb_outbox ( id UNIQUEIDENTIFIER PRIMARY KEY DEFAULT NEWID(), topic NVARCHAR(255) NOT NULL, [key] NVARCHAR(255), aggregate_type NVARCHAR(255), body NVARCHAR(MAX) NOT NULL, headers NVARCHAR(MAX) NOT NULL DEFAULT '{}', state NVARCHAR(50) NOT NULL DEFAULT 'pending', attempt INT NOT NULL DEFAULT 0, max_attempts INT NOT NULL DEFAULT 5, scheduled_at DATETIME2 NOT NULL DEFAULT GETUTCDATE(), created_at DATETIME2 NOT NULL DEFAULT GETUTCDATE(), updated_at DATETIME2 NOT NULL DEFAULT GETUTCDATE(), claimed_at DATETIME2 NULL, claim_token UNIQUEIDENTIFIER NULL, lease_expires_at DATETIME2 NULL, last_error NVARCHAR(MAX) NULL, sequence BIGINT IDENTITY(1,1) NOT NULL ); CREATE INDEX qb_outbox_pending_scheduled ON qb_outbox (state, scheduled_at) WHERE state = 'pending'; CREATE INDEX qb_outbox_state_updated ON qb_outbox (state, updated_at); CREATE INDEX qb_outbox_processing_claimed ON qb_outbox (claimed_at) WHERE state = 'processing'; CREATE INDEX qb_outbox_key_sequence ON qb_outbox ([key], sequence) WHERE state IN ('pending', 'processing'); CREATE TABLE qb_inbox ( id UNIQUEIDENTIFIER PRIMARY KEY DEFAULT NEWID(), source NVARCHAR(255) NOT NULL, dedup_key NVARCHAR(255) NOT NULL, aggregate_id NVARCHAR(255), event_type NVARCHAR(255), payload NVARCHAR(MAX) NOT NULL, headers NVARCHAR(MAX) NOT NULL DEFAULT '{}', state NVARCHAR(50) NOT NULL DEFAULT 'pending', consumption VARCHAR(4) NOT NULL DEFAULT 'push' CHECK (consumption IN ('push', 'pull')), attempt INT NOT NULL DEFAULT 0, last_error NVARCHAR(MAX) NULL, scheduled_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(), created_at DATETIME2 NOT NULL DEFAULT GETUTCDATE(), processed_at DATETIME2 NULL, claimed_at DATETIME2 NULL, claim_token UNIQUEIDENTIFIER NULL, lease_expires_at DATETIME2 NULL, correlation_id NVARCHAR(128) NULL, CONSTRAINT qb_inbox_source_idempotency UNIQUE (source, dedup_key) ); CREATE INDEX qb_inbox_state ON qb_inbox (state); CREATE INDEX qb_inbox_state_created ON qb_inbox (state, created_at); CREATE INDEX qb_inbox_aggregate_state ON qb_inbox (aggregate_id, state); CREATE INDEX qb_inbox_processing_claimed ON qb_inbox (claimed_at) WHERE state = 'processing'; CREATE INDEX qb_inbox_consumption_pending ON qb_inbox (consumption, state, scheduled_at); CREATE INDEX qb_inbox_pull_pending ON qb_inbox (source, scheduled_at, created_at, id) WHERE consumption = 'pull' AND state = 'pending'; CREATE INDEX qb_inbox_pull_busy ON qb_inbox (source, aggregate_id, lease_expires_at) WHERE consumption = 'pull' AND state = 'processing'; ``` Caution Keep the unique constraint on the source and idempotency key columns. QueueBox deduplicates the inbox through that constraint. Without it, the inbox stores every repeat. For change data capture, also create `queuebox_capture_state` from `V7__capture_state.sql`. Its name is fixed. ### Tell the pull clients [Section titled “Tell the pull clients”](#tell-the-pull-clients) A pull client builds its own SQL, so it needs the same names. Pass them through the schema option of the library: * Go ```go schema := queuebox.DefaultSchema() schema.Table = "qb_inbox" schema.IdempotencyKey = "dedup_key" worker, err := queuebox.NewInboxWorker(db, queuebox.Options{Source: "orders", Schema: &schema}) ``` * TypeScript ```ts import { defaultSchema } from "@alternayte/queuebox-inbox"; const worker = new InboxWorker(fromPg(pool), { source: "orders", schema: { ...defaultSchema, table: "qb_inbox", idempotencyKey: "dedup_key" }, }); ``` * C# ```csharp var options = new InboxOptions { Source = "orders", Schema = InboxSchema.Default with { Table = "qb_inbox", IdempotencyKey = "dedup_key" }, }; ``` The library quotes and checks every name, so a mapping cannot carry SQL. # Use QueueBox from an EF Core app > Write outbox rows with SaveChanges and consume the inbox with a DbContext that shares the handler transaction. This guide shows you how an Entity Framework Core application writes outbox rows and consumes inbox messages. Both sides follow one rule: the QueueBox row and the business write share one transaction on one connection. ## Map the outbox table [Section titled “Map the outbox table”](#map-the-outbox-table) QueueBox owns and migrates the `outbox` table. Map an entity to it, but do not generate a migration from the entity. ```csharp public class OutboxMessage { public Guid Id { get; set; } public string Topic { get; set; } = default!; public string? Key { get; set; } public string Payload { get; set; } = default!; public string Headers { get; set; } = "{}"; public string? AggregateType { get; set; } } ``` * PostgreSQL ```csharp public class OutboxMessageConfiguration : IEntityTypeConfiguration { public void Configure(EntityTypeBuilder builder) { builder.ToTable("outbox", t => t.ExcludeFromMigrations()); builder.Property(m => m.Id).HasColumnName("id"); builder.Property(m => m.Topic).HasColumnName("topic"); builder.Property(m => m.Key).HasColumnName("key"); builder.Property(m => m.Payload).HasColumnName("payload").HasColumnType("jsonb"); builder.Property(m => m.Headers).HasColumnName("headers").HasColumnType("jsonb"); builder.Property(m => m.AggregateType).HasColumnName("aggregate_type"); } } ``` * SQL Server ```csharp public class OutboxMessageConfiguration : IEntityTypeConfiguration { public void Configure(EntityTypeBuilder builder) { builder.ToTable("outbox", t => t.ExcludeFromMigrations()); builder.Property(m => m.Id).HasColumnName("id"); builder.Property(m => m.Topic).HasColumnName("topic"); builder.Property(m => m.Key).HasColumnName("key"); builder.Property(m => m.Payload).HasColumnName("payload").HasColumnType("nvarchar(max)"); builder.Property(m => m.Headers).HasColumnName("headers").HasColumnType("nvarchar(max)"); builder.Property(m => m.AggregateType).HasColumnName("aggregate_type").HasColumnType("nvarchar(255)"); } } ``` - Name every column. The default convention does not turn `AggregateType` into `aggregate_type`. - Map only the columns that the application writes. The database fills `state`, `attempt`, `scheduled_at`, `created_at`, `updated_at` and `sequence` with their defaults. See [the outbox table](/reference/outbox-table/). - Set `Headers` to a JSON object string, never to `null`. The column is `NOT NULL`. ## Write a row with the business change [Section titled “Write a row with the business change”](#write-a-row-with-the-business-change) Add the business entity and the outbox row to one context, then call `SaveChanges` once. `SaveChanges` wraps both inserts in one transaction, so they commit together or not at all. ```csharp public async Task CreateOrder(ShopContext db, Guid orderId, string customerId, decimal amount, CancellationToken token) { db.Orders.Add(new Order { Id = orderId, CustomerId = customerId, Amount = amount }); db.OutboxMessages.Add(new OutboxMessage { Id = Guid.NewGuid(), Topic = "order.created", Key = customerId, Payload = JsonSerializer.Serialize(new { orderId, amount }), Headers = """{"X-Tenant":"acme"}""", AggregateType = "order", }); await db.SaveChangesAsync(token); } ``` When the work needs more than one `SaveChanges`, open the transaction yourself with `db.Database.BeginTransactionAsync` and commit it after the last call. Set `Key` to the ID of the thing whose messages must arrive in order, for example the customer or the order. QueueBox delivers the rows of one key in insert order, one at a time. See [ordering](/concepts/ordering/#order-and-the-key). Note An event-sourced application on [Deedbox](https://deedbox-docs.pages.dev) does not need this entity. The `Deedbox.QueueBox` package writes the outbox row in the transaction of each append and sets `key` to the stream ID. See [Wire QueueBox](https://deedbox-docs.pages.dev/how-to/wire-queuebox/). ## Consume the inbox with a DbContext [Section titled “Consume the inbox with a DbContext”](#consume-the-inbox-with-a-dbcontext) A pull worker from `QueueBox.Inbox.DependencyInjection` claims inbox messages and gives each handler an open transaction. The handler writes through that transaction. The library completes the message in the same transaction and commits it. ```sh dotnet add package QueueBox.Inbox.DependencyInjection ``` Build the context with `InboxDbContextFactory.CreateOn`. It hands the handler connection to your `build` delegate and enlists the context in the handler transaction. * PostgreSQL ```csharp using Npgsql; using QueueBox.Inbox; using QueueBox.Inbox.DependencyInjection; var builder = Host.CreateApplicationBuilder(args); builder.Services.AddSingleton( InboxConnections.From(NpgsqlDataSource.Create(builder.Configuration.GetConnectionString("Queuebox")))); builder.Services.AddQueueBoxInbox("payments", new InboxOptions { Source = "payments" }, async (message, transaction, token) => { await using var db = InboxDbContextFactory.CreateOn( transaction, connection => new ShopContext(new DbContextOptionsBuilder().UseNpgsql(connection).Options)); var orderId = message.Payload.GetProperty("orderId").GetGuid(); var order = await db.Orders.SingleAsync(o => o.Id == orderId, token); order.PaidAt = DateTimeOffset.UtcNow; await db.SaveChangesAsync(token); }); await builder.Build().RunAsync(); ``` * SQL Server ```csharp using Microsoft.Data.SqlClient; using QueueBox.Inbox; using QueueBox.Inbox.DependencyInjection; var builder = Host.CreateApplicationBuilder(args); builder.Services.AddSingleton( InboxConnections.From(SqlClientFactory.Instance, builder.Configuration.GetConnectionString("Queuebox")!)); builder.Services.AddQueueBoxInbox( "payments", new InboxOptions { Source = "payments", Dialect = SqlDialect.SqlServer }, async (message, transaction, token) => { await using var db = InboxDbContextFactory.CreateOn( transaction, connection => new ShopContext(new DbContextOptionsBuilder().UseSqlServer(connection).Options)); var orderId = message.Payload.GetProperty("orderId").GetGuid(); var order = await db.Orders.SingleAsync(o => o.Id == orderId, token); order.PaidAt = DateTimeOffset.UtcNow; await db.SaveChangesAsync(token); }); await builder.Build().RunAsync(); ``` The inbox source must use pull consumption. See [Consume the inbox](/how-to/consume-the-inbox/) for the source configuration. Follow these rules in a handler: 1. Build every `DbContext` through `InboxDbContextFactory.CreateOn`. A context from dependency injection or from `new` opens its own connection. Its write then commits on its own, and a later failure makes QueueBox deliver the message again after the write already exists. 2. Do not call `BeginTransaction`, `Commit` or `Rollback`. The library owns the transaction. 3. Throw to fail the message. The library rolls back every write of the handler. 4. Pass the cancellation token on. It fires when the lease is lost. 5. Make a call to another system safe to repeat. Deduplicate it on the source and the idempotency key. `InboxDbContextFactory` references only `Microsoft.EntityFrameworkCore.Relational`. Add the provider package that your `build` delegate uses: `Npgsql.EntityFrameworkCore.PostgreSQL` or `Microsoft.EntityFrameworkCore.SqlServer`. ## Next steps [Section titled “Next steps”](#next-steps) * [Write outbox rows](/how-to/write-outbox-rows/) holds the SQL contract that the entity above writes. * [Pull clients](/reference/pull-clients/#c) lists every option of the C# client. * [Delivery semantics](/concepts/delivery-semantics/) states what QueueBox promises for each message. # Write outbox rows > Insert an outbox row in the transaction of your business write, and set its key, headers, schedule and retry ceiling. This guide shows how an application hands a message to QueueBox. The application runs one `INSERT` into the `outbox` table, inside the transaction of its business write. QueueBox reads the table, routes each row to a destination, and delivers it. [The outbox table](/reference/outbox-table/) lists every column, its type and its default. This guide shows the columns an application writes. Note `app/src/test/kotlin/docs/IntegrationDocSqlTest.kt` runs every SQL block on this page. The PostgreSQL blocks run against a PostgreSQL container, and the SQL Server blocks run against a SQL Server container. The running poller then delivers every inserted row to an HTTP destination. ## The insert shares the business transaction [Section titled “The insert shares the business transaction”](#the-insert-shares-the-business-transaction) Open one transaction, write the business rows, insert the outbox row, then commit. This shared transaction is the reason for the outbox pattern. The database commits the message and the business state together, so the two never disagree. * If the transaction commits, the business row and the outbox row both exist. QueueBox then delivers the message at least once. * If the transaction rolls back, neither row exists. No message announces an order that does not exist. An application that publishes to a broker outside the transaction has no such guarantee. The commit can fail after the publish, and the message then announces an order that does not exist. The publish can fail after the commit, and the message is lost. No order of the two writes removes the failure, because they go to two systems. Two more rules follow from this one: * Do not open a second connection for the outbox insert. A second connection is a second transaction, and the guarantee is gone. * Keep the transaction short. QueueBox sees the row only after the commit. ## Insert a row [Section titled “Insert a row”](#insert-a-row) The examples use one business table, `orders`. The examples create it, because the test runs every statement. Your application has its own table already. * PostgreSQL ```sql CREATE TABLE IF NOT EXISTS orders ( id UUID PRIMARY KEY, customer_id VARCHAR(64) NOT NULL, amount NUMERIC(12, 2) NOT NULL, created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP ); ``` `payload` and `headers` are `JSONB`, so cast each string literal. ```sql BEGIN; INSERT INTO orders (id, customer_id, amount) VALUES ('11111111-1111-1111-1111-111111111111', 'cust-42', 99.99); INSERT INTO outbox (topic, key, payload, headers, aggregate_type) VALUES ( 'order.created', 'cust-42', '{"orderId":"11111111-1111-1111-1111-111111111111","amount":99.99}'::jsonb, '{"X-Tenant":"acme"}'::jsonb, 'order' ); COMMIT; ``` * SQL Server ```sql IF OBJECT_ID('orders', 'U') IS NULL CREATE TABLE orders ( id UNIQUEIDENTIFIER PRIMARY KEY, customer_id NVARCHAR(64) NOT NULL, amount DECIMAL(12, 2) NOT NULL, created_at DATETIME2 NOT NULL DEFAULT GETUTCDATE() ); ``` `payload` and `headers` are `NVARCHAR(MAX)`. `key` is a reserved word, so put it in brackets. ```sql BEGIN TRANSACTION; INSERT INTO orders (id, customer_id, amount) VALUES ('11111111-1111-1111-1111-111111111111', N'cust-42', 99.99); INSERT INTO outbox (topic, [key], payload, headers, aggregate_type) VALUES ( N'order.created', N'cust-42', N'{"orderId":"11111111-1111-1111-1111-111111111111","amount":99.99}', N'{"X-Tenant":"acme"}', N'order' ); COMMIT TRANSACTION; ``` The insert names five columns: | Column | What to write | | ---------------- | ----------------------------------------------------------------------------------------- | | `topic` | Required. The name that routes the row, for example `order.created`. | | `payload` | Required. The message body, as a JSON object. | | `key` | Optional. The unit of order. See [Key and order](#key-and-order). | | `headers` | Optional. A JSON object of headers that travel with the message. See [Headers](#headers). | | `aggregate_type` | Optional. The kind of business entity, for example `order`. | `aggregate_type` lets a broker destination choose its address from the row. A RabbitMQ exchange, a Kafka topic or a NATS subject can be a template such as `public.orders.{{ aggregateType }}.v1`. The destination can also read the column through `exchangeFrom`, `topicFrom` or `subjectFrom`. See [Bridge a broker](/how-to/bridge-a-broker/). The shortest legal insert names only `topic` and `payload`. Every other column has a default or accepts null. * PostgreSQL ```sql BEGIN; INSERT INTO orders (id, customer_id, amount) VALUES ('22222222-2222-2222-2222-222222222222', 'cust-7', 12.00); INSERT INTO outbox (topic, payload) VALUES ('order.paid', '{"orderId":"22222222-2222-2222-2222-222222222222"}'::jsonb); COMMIT; ``` * SQL Server ```sql BEGIN TRANSACTION; INSERT INTO orders (id, customer_id, amount) VALUES ('22222222-2222-2222-2222-222222222222', N'cust-7', 12.00); INSERT INTO outbox (topic, payload) VALUES (N'order.paid', N'{"orderId":"22222222-2222-2222-2222-222222222222"}'); COMMIT TRANSACTION; ``` ### Columns the application must not write [Section titled “Columns the application must not write”](#columns-the-application-must-not-write) QueueBox owns `state`, `attempt`, `claimed_at`, `claim_token`, `lease_expires_at` and `last_error`. The database fills `sequence`. Write `state` only with the value `pending`, which is the default. A row that starts in another state can stall, or it can deliver twice. ## Choose a topic [Section titled “Choose a topic”](#choose-a-topic) QueueBox matches `topic` against the `topicPattern` of each route, in configuration order. The first route that matches wins. A topic that matches no route goes to `dead`. Use a dotted, lower case topic, for example `order.created`. The column holds 255 characters at most. [Fan out over HTTP](/how-to/fan-out-over-http/#match-topics-with-patterns) shows how a pattern matches a topic. ## Key and order [Section titled “Key and order”](#key-and-order) The rows of one non-empty `key` arrive in insert order, one row at a time. Rows with a null or empty `key`, and rows of different keys, have no order between them. Use the identifier of the entity whose events must stay in order, for example the order ID or the customer ID. The two rows below share a key, so QueueBox delivers `order.created` before `order.paid`. * PostgreSQL ```sql BEGIN; INSERT INTO outbox (topic, key, payload, aggregate_type) VALUES ( 'order.created', 'order-33333333', '{"orderId":"33333333-3333-3333-3333-333333333333","step":1}'::jsonb, 'order' ); INSERT INTO outbox (topic, key, payload, aggregate_type) VALUES ( 'order.paid', 'order-33333333', '{"orderId":"33333333-3333-3333-3333-333333333333","step":2}'::jsonb, 'order' ); COMMIT; ``` * SQL Server ```sql BEGIN TRANSACTION; INSERT INTO outbox (topic, [key], payload, aggregate_type) VALUES ( N'order.created', N'order-33333333', N'{"orderId":"33333333-3333-3333-3333-333333333333","step":1}', N'order' ); INSERT INTO outbox (topic, [key], payload, aggregate_type) VALUES ( N'order.paid', N'order-33333333', N'{"orderId":"33333333-3333-3333-3333-333333333333","step":2}', N'order' ); COMMIT TRANSACTION; ``` These rules decide what a key costs and what it gives: * The database fills `sequence` on insert, and the claim orders the rows of one key by it. Two rows that one transaction writes keep their insert order. * A row that waits for a retry holds back the later rows of its key. A dead row releases its key. * One row of a key is in flight at a time. The throughput of one key is one publish round trip per row. * The order holds for one writer per key. Two transactions that insert rows of one key at the same time can commit out of `sequence` order. Spread a busy stream over more keys only when the consumer needs no order across them. [Ordering](/concepts/ordering/#order-and-the-key) explains the rule in full. ## Headers [Section titled “Headers”](#headers) `headers` holds one JSON object. Each key is a header name, and each value is a string. The column is `NOT NULL` with the default `'{}'`. * To send no header, omit the column. An explicit `NULL` makes the insert fail. * An HTTP destination sends each row header as an HTTP header. A row header wins over a static destination header and over the authentication header of the same name. * A broker destination sends the row headers as message headers. * Put `X-Correlation-Id` in `headers` to follow one message through every log line. [Headers](/reference/headers/) lists the headers that QueueBox adds itself. ## Delay a message [Section titled “Delay a message”](#delay-a-message) Set `scheduled_at` to delay a message. QueueBox claims a row only when `scheduled_at` has passed. The default is the moment of the insert. `max_attempts` sets the retry ceiling of one row. The block below delays a reminder by five seconds and gives it ten attempts. * PostgreSQL ```sql BEGIN; INSERT INTO outbox (topic, payload, scheduled_at, max_attempts) VALUES ( 'order.reminder', '{"orderId":"22222222-2222-2222-2222-222222222222"}'::jsonb, CURRENT_TIMESTAMP + INTERVAL '5 seconds', 10 ); COMMIT; ``` * SQL Server ```sql BEGIN TRANSACTION; INSERT INTO outbox (topic, payload, scheduled_at, max_attempts) VALUES ( N'order.reminder', N'{"orderId":"22222222-2222-2222-2222-222222222222"}', DATEADD(SECOND, 5, SYSUTCDATETIME()), 10 ); COMMIT TRANSACTION; ``` On SQL Server, `scheduled_at` is `DATETIME2` and carries no time zone. QueueBox compares it with the wall clock of the host that runs QueueBox. Run every QueueBox instance in UTC, and write UTC values. ## Set the retry ceiling [Section titled “Set the retry ceiling”](#set-the-retry-ceiling) QueueBox compares the `attempt` column of a row with the `max_attempts` column of the same row. `attempt` is `0` on the first delivery, and each failed delivery raises it. The ceiling comes from the first of these that applies: 1. The `max_attempts` value that the application writes on the row. 2. `outbox.maxAttempts`, which QueueBox writes on every row it creates itself, for example a row that the inbox relay forwards. `inbox.relay.maxAttempts` overrides it for relayed rows. 3. The column default, `5`. Set `max_attempts` on the insert to give a slow destination more attempts than the rest of the system. A row that reaches its ceiling goes to `dead`. See [Dead letters](/operations/dead-letters/). ## Write the row from application code [Section titled “Write the row from application code”](#write-the-row-from-application-code) Any database library can write the row, because the row is one insert into one table. QueueBox publishes no library for the outbox side. Check two things in your code: 1. The insert joins the transaction of the business write. 2. `payload` and `headers` hold a JSON object. A JSON string that holds another JSON string breaks a transform and a routing key template. * Go ```go func createOrder(ctx context.Context, db *sql.DB, order Order) error { tx, err := db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() if _, err := tx.ExecContext(ctx, "INSERT INTO orders (id, customer_id, amount) VALUES ($1, $2, $3)", order.ID, order.CustomerID, order.Amount); err != nil { return err } payload, err := json.Marshal(map[string]any{"orderId": order.ID, "amount": order.Amount}) if err != nil { return err } if _, err := tx.ExecContext(ctx, `INSERT INTO outbox (topic, key, payload, headers, aggregate_type) VALUES ($1, $2, $3::jsonb, $4::jsonb, $5)`, "order.created", order.CustomerID, string(payload), `{"X-Tenant":"acme"}`, "order"); err != nil { return err } return tx.Commit() } ``` * TypeScript ```ts import pg from "pg"; const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL }); export async function createOrder(order: { id: string; customerId: string; amount: number }) { const client = await pool.connect(); try { await client.query("BEGIN"); await client.query( "INSERT INTO orders (id, customer_id, amount) VALUES ($1, $2, $3)", [order.id, order.customerId, order.amount], ); await client.query( "INSERT INTO outbox (topic, key, payload, headers, aggregate_type) VALUES ($1, $2, $3, $4, $5)", [ "order.created", order.customerId, JSON.stringify({ orderId: order.id, amount: order.amount }), JSON.stringify({ "X-Tenant": "acme" }), "order", ], ); await client.query("COMMIT"); } catch (error) { await client.query("ROLLBACK"); throw error; } finally { client.release(); } } ``` * C# ```csharp var orderId = Guid.NewGuid(); await using var transaction = await db.Database.BeginTransactionAsync(cancellationToken); db.Orders.Add(new Order { Id = orderId, CustomerId = "cust-42", Amount = 99.99m }); db.OutboxMessages.Add(new OutboxMessage { Id = Guid.NewGuid(), Topic = "order.created", Key = "cust-42", Payload = JsonSerializer.Serialize(new { orderId, amount = 99.99m }), Headers = """{"X-Tenant":"acme"}""", AggregateType = "order", }); await db.SaveChangesAsync(cancellationToken); await transaction.CommitAsync(cancellationToken); ``` The next section holds the `OutboxMessage` entity and its mapping. ### The Entity Framework Core mapping [Section titled “The Entity Framework Core mapping”](#the-entity-framework-core-mapping) Entity Framework Core writes the properties of an entity, not the columns of a table. An entity without a `Headers` property or an `AggregateType` property never writes those columns. The default convention also does not turn `AggregateType` into `aggregate_type`, so name every column explicitly. ```csharp public class OutboxMessage { public Guid Id { get; set; } public string Topic { get; set; } = default!; public string? Key { get; set; } public string Payload { get; set; } = default!; public string Headers { get; set; } = "{}"; public string? AggregateType { get; set; } } public class OutboxMessageConfiguration : IEntityTypeConfiguration { public void Configure(EntityTypeBuilder builder) { builder.ToTable("outbox"); builder.Property(m => m.Id).HasColumnName("id"); builder.Property(m => m.Topic).HasColumnName("topic"); builder.Property(m => m.Key).HasColumnName("key"); builder.Property(m => m.Payload).HasColumnName("payload").HasColumnType("jsonb"); builder.Property(m => m.Headers).HasColumnName("headers").HasColumnType("jsonb"); builder.Property(m => m.AggregateType).HasColumnName("aggregate_type"); } } ``` * `HasColumnType("jsonb")` applies to PostgreSQL only. On SQL Server, use `nvarchar(max)` for `Payload` and `Headers`, and `nvarchar(255)` for `AggregateType`. * Set `Headers` to a JSON object string, never to `null`. The column is `NOT NULL`. * Map `AggregateType` even when no destination reads it today. An unmapped property leaves the column empty for a destination that reads it later. * QueueBox owns and migrates the `outbox` table. Do not generate an Entity Framework Core migration from this entity. [Use QueueBox from an EF Core app](/how-to/use-entity-framework-core/) shows the SQL Server mapping, and the inbox side with `InboxDbContextFactory`. ## Next steps [Section titled “Next steps”](#next-steps) * [Fan out over HTTP](/how-to/fan-out-over-http/) routes the rows to HTTP endpoints. * [Bridge a broker](/how-to/bridge-a-broker/) publishes the rows to RabbitMQ, Kafka or NATS. * [Delivery semantics](/concepts/delivery-semantics/) states what QueueBox promises for a committed row. # Dead letters > List, requeue and discard dead outbox and inbox messages with SQL. This page gives the supported SQL to list, requeue and discard a dead message. To replay through the HTTP API with filters, see [Replay dead letters](/how-to/replay-dead-letters/). ## When a message is dead [Section titled “When a message is dead”](#when-a-message-is-dead) An outbox message reaches the state `dead` when: * its `attempt` count reaches the `max_attempts` of its own row, * no route matches its topic, * no publisher supports its destination, * or a transform with `onError: Dead` rejects it. QueueBox writes `outbox.maxAttempts` into `max_attempts` for every row that it creates. Your application can set a different value on a row that it inserts, and the row value wins. See [Configuration](/reference/configuration/). An inbox message reaches the state `dead` when: * an AMQP source stores a message that the source transform rejected, * a broker source receives a body that is not JSON, * the topic template of a push source renders empty, * or a pull worker gives up on it. QueueBox never deletes a dead message on its own. The retention job removes it after `retention.outbox.maxAge` or `retention.inbox.maxAge`, when retention is on. Watch `queuebox_outbox_messages_total{status="dead"}` to see new dead messages. ## How to read the SQL [Section titled “How to read the SQL”](#how-to-read-the-sql) The SQL is for the shipped PostgreSQL schema. A test runs every `sql` block on this page against that schema. Another test runs the `requeue-one` block and checks that the destination then receives the message. * Each statement ends with a semicolon at the end of a line. * A placeholder starts with a colon, for example `:message_id`. Replace it with a real value before you run the statement. * The placeholders are `:message_id`, `:topic`, `:state`, `:destination` and `:limit`. * An MDX comment `{/* sql-id: name */}` before a block gives the block a name. A test selects the block by that name. On SQL Server, write `SELECT TOP (n)` in place of `LIMIT n`, and `N'text'` for a string literal. The `UPDATE` and `DELETE` statements run unchanged. ## List the dead messages [Section titled “List the dead messages”](#list-the-dead-messages) Count the dead messages per topic: ```sql SELECT topic, count(*) AS dead_count FROM outbox WHERE state = 'dead' GROUP BY topic ORDER BY dead_count DESC; ``` List the newest dead messages with the failure reason: ```sql SELECT id, topic, key, attempt, max_attempts, created_at, updated_at, last_error FROM outbox WHERE state = 'dead' ORDER BY updated_at DESC LIMIT :limit; ``` Read one dead message in full, with its payload and its headers: ```sql SELECT id, topic, key, payload, headers, attempt, max_attempts, scheduled_at, created_at, updated_at, claimed_at, last_error FROM outbox WHERE id = :message_id AND state = 'dead'; ``` `last_error` holds the reason for the last failure. QueueBox masks secret values in it and truncates it. ## Requeue one dead message [Section titled “Requeue one dead message”](#requeue-one-dead-message) Correct the cause of the failure first. A requeue against a destination that is still broken produces a second dead message. The requeue sets the state to `pending`, resets `attempt` to zero and sets `scheduled_at` to the current time. The poller claims the row on its next cycle. ```sql UPDATE outbox SET state = 'pending', attempt = 0, scheduled_at = CURRENT_TIMESTAMP, claimed_at = NULL, last_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = :message_id AND state = 'dead'; ``` The clause `AND state = 'dead'` protects a message that another operator already requeued. The statement then reports zero updated rows. Confirm the new state: ```sql SELECT id, state, attempt, scheduled_at, last_error FROM outbox WHERE id = :message_id; ``` The message is delivered when its state becomes `sent`. Note A requeued row keeps its `sequence`. If the row has a `key`, it becomes the head of its key again and holds back the later rows of that key until it is sent or dead. It reaches the destination after the rows of its key that passed it while it was dead. See [Ordering](/concepts/ordering/#order-and-the-key). ## Requeue every dead message of one topic [Section titled “Requeue every dead message of one topic”](#requeue-every-dead-message-of-one-topic) Use this form after you repair a destination. Requeue one message first and confirm the delivery. ```sql UPDATE outbox SET state = 'pending', attempt = 0, scheduled_at = CURRENT_TIMESTAMP, claimed_at = NULL, last_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE state = 'dead' AND topic = :topic; ``` ## Requeue a dead inbox message [Section titled “Requeue a dead inbox message”](#requeue-a-dead-inbox-message) List the dead inbox rows first: ```sql SELECT id, source, consumption, idempotency_key, event_type, attempt, created_at, last_error FROM inbox WHERE state = 'dead' ORDER BY created_at DESC LIMIT :limit; ``` Correct the cause first. For a transform rejection, correct the source transform. For an empty topic, set `sources..topic` or the event type path. The requeue sets the state back to `pending`. The relay then forwards a push row, and a pull worker claims a pull row. The requeue also resets `attempt`, so a pull worker applies its full retry ceiling again. ```sql UPDATE inbox SET state = 'pending', attempt = 0, scheduled_at = CURRENT_TIMESTAMP, last_error = NULL, claimed_at = NULL, processed_at = NULL WHERE id = :message_id AND state = 'dead'; ``` Caution A requeued inbox row keeps its stored payload. A payload that the source transform rejected at receipt was stored untransformed, and the relay runs no transform. Correct the payload in the row, or have the sender send a corrected message under a new idempotency key. ## Discard a dead message [Section titled “Discard a dead message”](#discard-a-dead-message) Delete a message only when you accept the loss of the event. Copy the payload first. ```sql DELETE FROM outbox WHERE id = :message_id AND state = 'dead'; ``` Deleting a dead inbox row also ends deduplication for its `(source, idempotency_key)`. A repeat of that message then becomes a new row. # Deploy QueueBox > Run the published image with Docker, Compose or Kubernetes, against a PostgreSQL or SQL Server database. This page tells you how to run QueueBox in production. It covers the image, the database, the configuration, the health probes and replicas. ## The image [Section titled “The image”](#the-image) QueueBox publishes a multi-architecture image for `linux/amd64` and `linux/arm64` to GitHub Container Registry. | Tag | Meaning | | ------------------------------------ | --------------------------------------- | | `ghcr.io/alternayte/queuebox:1.2.3` | The exact release. | | `ghcr.io/alternayte/queuebox:1.2` | The newest patch of that minor release. | | `ghcr.io/alternayte/queuebox:latest` | The newest release. | Pin the exact version in production. The registry also holds the software bill of materials and the build provenance of each image. The image runs Java 21 as the non-root user `queuebox`. It listens on port 8080. Its built-in `HEALTHCHECK` calls `http://localhost:8080/health`. Run the image against an existing database: ```bash docker run --rm -p 8080:8080 \ -e QUEUEBOX_DATABASE_URL=jdbc:postgresql://db.internal:5432/queuebox \ -e QUEUEBOX_DATABASE_USERNAME=queuebox \ -e QUEUEBOX_DATABASE_PASSWORD=secret \ -v /etc/queuebox/queuebox.yml:/etc/queuebox/queuebox.yml:ro \ ghcr.io/alternayte/queuebox:1.2.3 ``` ## The database [Section titled “The database”](#the-database) QueueBox needs one of these databases: * PostgreSQL 14, 15 or 16. * SQL Server 2019 or 2022. Set `database.type` to `postgresql` (the default) or `sqlserver`. At start, QueueBox waits up to `database.startupTimeoutMs` (default 60000) for the database. It exits when no database answers in that time. ### Migrations [Section titled “Migrations”](#migrations) By default, QueueBox applies its bundled Flyway migrations at start. The database account then needs the right to create and alter tables, indexes and sequences. Set `database.migrate: false` when the application account has no DDL rights. Apply the migration files by hand before each upgrade. The files are in [`postgres/src/main/resources/db/postgresql`](https://github.com/alternayte/queuebox/blob/main/postgres/src/main/resources/db/postgresql) and [`sqlserver/src/main/resources/db/sqlserver`](https://github.com/alternayte/queuebox/blob/main/sqlserver/src/main/resources/db/sqlserver). A custom table name or column name requires `database.migrate: false`. QueueBox refuses to start with a renamed table or column and `migrate: true`, because the bundled files name the default schema. See [Use custom tables](/how-to/use-custom-tables/). At start, QueueBox also checks that the tables have the columns it needs. A missing required column stops the start with a message that gives the `ALTER TABLE` statement. ### Connections [Section titled “Connections”](#connections) Each replica opens up to `database.poolSize` connections (default 10). The poller, the relay, the retention job and the HTTP routes share the pool. Keep `database.poolSize` larger than `outbox.concurrency`. Check that the database allows `poolSize` times the number of replicas, plus your other clients. ## The configuration [Section titled “The configuration”](#the-configuration) QueueBox reads its configuration from three places. The first place that holds a key wins. 1. An environment variable that starts with `QUEUEBOX_`. 2. The YAML file that `QUEUEBOX_CONFIG_FILE` names, or `/etc/queuebox/queuebox.yml` when the variable is not set. 3. The YAML file packaged in the image. An external file **replaces** the packaged file. It does not add to it. Write a complete configuration, and copy [`examples/queuebox.yml`](https://github.com/alternayte/queuebox/blob/main/examples/queuebox.yml) as the start point. QueueBox reads the packaged file only when you supply no external file and set no `QUEUEBOX_` variable. These three variables are the minimum: ```bash QUEUEBOX_DATABASE_URL=jdbc:postgresql://db.internal:5432/queuebox QUEUEBOX_DATABASE_USERNAME=queuebox QUEUEBOX_DATABASE_PASSWORD=secret ``` The name after the prefix is the configuration path in upper case. One underscore separates two levels of the path. A leaf name of more than one word has no underscore inside it: `QUEUEBOX_OUTBOX_POLLINTERVALMS` sets `outbox.pollIntervalMs`. A variable with an extra underscore sets nothing, and the start reports no error. See [Environment variables](/reference/environment-variables/) and [Configuration](/reference/configuration/). A configuration change needs a restart. It does not need a new image. Keep secrets out of the YAML file and out of plain environment values. See [Security](/operations/security/#secrets). ## Docker Compose [Section titled “Docker Compose”](#docker-compose) The repository holds a Compose stack with QueueBox, PostgreSQL and a test receiver: ```bash git clone https://github.com/alternayte/queuebox.git cd queuebox docker compose -f docker-compose.yml --env-file .env.example up -d --build ``` * `-f docker-compose.yml` selects the shipped stack. Without it, Compose also reads `docker-compose.override.yml`, which runs a development loop from the source tree. * The stack mounts `examples/queuebox.yml` at `/etc/queuebox/queuebox.yml`. Edit that file and restart the container. * `--env-file .env.example` supplies the database variables. Without it, the start fails with a message that names the variables to set. * `docker compose --profile rabbitmq up -d` also starts RabbitMQ. The stack builds the image from the source. To run a released image with it, add `docker-compose.release.yml`: ```bash RELEASE_IMAGE=ghcr.io/alternayte/queuebox:1.2.3 \ docker compose -f docker-compose.yml -f docker-compose.release.yml --env-file .env.example up -d ``` Use this stack to learn and to test. For production, run the image against a database that you back up and monitor. ## Run without a container [Section titled “Run without a container”](#run-without-a-container) This section runs QueueBox from the source against a PostgreSQL database on the same host. It needs a Java Development Kit, version 21. CI runs these steps on every push to `main`. 1. Create the database: ```sql CREATE DATABASE queuebox; ``` QueueBox creates its tables at startup with the bundled migrations. 2. Copy the example configuration. Do not edit `config/src/main/resources/queuebox.yml`, because the image packages that file. ```bash sudo mkdir -p /etc/queuebox sudo cp examples/queuebox.yml /etc/queuebox/queuebox.yml ``` 3. Set the database variables. An environment variable wins over every file. ```bash export QUEUEBOX_DATABASE_URL=jdbc:postgresql://localhost:5432/queuebox export QUEUEBOX_DATABASE_USERNAME=postgres export QUEUEBOX_DATABASE_PASSWORD=secret ``` 4. Build and run: ```bash ./gradlew run ``` 5. Confirm that the instance is ready. `GET /health` answers `200` when every component is up. ```bash curl http://localhost:8080/health ``` ## Health probes [Section titled “Health probes”](#health-probes) | Endpoint | Use | Answer | | ------------------- | --------------- | --------------------------------------------------------------- | | `GET /health/live` | Liveness probe | Always 200 while the process runs. It touches no dependency. | | `GET /health/ready` | Readiness probe | 200 when every component is up. 503 when one component is down. | | `GET /health` | Compatibility | The same answer as `/health/ready`. | The readiness body names each component: * `database`, * `outbox-poller`, * `inbox-relay`, when `inbox.relay.enabled` is true, * `rabbitmq.`, `kafka.` and `nats.` for each broker source, * `retention-service`, when `retention.enabled` is true, * `outbox-capture`, when capture is enabled. This component is advisory: a capture fault never makes the answer 503. Each check has a 3-second limit. A check that does not answer in time counts as down, so a slow database gives 503 and not a probe timeout. Use `/health/live` for liveness. A liveness probe on `/health/ready` restarts the pod when the database is slow, and a restart does not repair the database. ### The management port [Section titled “The management port”](#the-management-port) Set `server.managementPort` to move `/health/*`, `/metrics` and `/admin` to a separate port. The data port then answers 404 for those paths and carries only the inbox routes. Bind the management port to an internal network, because the metrics reveal traffic volumes and destination names. Caution The image `HEALTHCHECK` calls `/health` on port 8080. With a management port, that call gets 404 and Docker marks the container unhealthy. Override the health check to call the management port. ## Kubernetes [Section titled “Kubernetes”](#kubernetes) QueueBox has no Helm chart. A plain Deployment and Service are enough. ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: queuebox spec: replicas: 2 selector: matchLabels: app: queuebox template: metadata: labels: app: queuebox spec: terminationGracePeriodSeconds: 75 containers: - name: queuebox image: ghcr.io/alternayte/queuebox:1.2.3 ports: - name: http containerPort: 8080 - name: management containerPort: 9090 env: - name: QUEUEBOX_CONFIG_FILE value: /etc/queuebox/queuebox.yml - name: QUEUEBOX_DATABASE_URL value: jdbc:postgresql://postgres.db.svc:5432/queuebox - name: QUEUEBOX_DATABASE_USERNAME value: queuebox - name: QUEUEBOX_DATABASE_PASSWORD value: file:/run/secrets/db-password - name: QUEUEBOX_SERVER_MANAGEMENTPORT value: "9090" livenessProbe: httpGet: path: /health/live port: management readinessProbe: httpGet: path: /health/ready port: management volumeMounts: - name: config mountPath: /etc/queuebox readOnly: true - name: secrets mountPath: /run/secrets readOnly: true volumes: - name: config configMap: name: queuebox-config - name: secrets secret: secretName: queuebox-secrets ``` * Put the complete `queuebox.yml` in the ConfigMap. * Mount secrets as files and reference them with `file:`. See [Security](/operations/security/#the-kubernetes-secret-pattern). * Publish only `/inbox` through the ingress, and terminate TLS there. See [Security](/operations/security/#transport-security). * Scrape `/metrics` on the management port. See [Monitoring](/operations/monitoring/). ### Shutdown [Section titled “Shutdown”](#shutdown) On `SIGTERM`, QueueBox stops in this order: 1. The HTTP server refuses new requests and waits up to 5 seconds for the requests in flight. 2. The broker consumers, capture, the outbox poller and the inbox relay stop. The poller waits up to `outbox.shutdownTimeoutMs` (default 30000) for the messages in flight. 3. The publishers and the database pool close. A message still in flight after the timeout stays in state `processing`. Another replica reclaims it when its lease expires. Set `terminationGracePeriodSeconds` above 5 seconds plus twice `outbox.shutdownTimeoutMs`. With the defaults, that is above 65 seconds. ## Replicas [Section titled “Replicas”](#replicas) Run as many replicas as you need. Every replica runs the outbox poller and the inbox relay against the same tables. Claims, leases and the claim fence keep two replicas off the same message, and the order of a key or an aggregate holds across replicas. See [Claims and leases](/concepts/claims-and-leases/). * Every replica needs the same configuration. A replica with a different route set routes the same topic differently. * Enable capture on one replica only. Set `outbox.capture.enabled: false` on the others. See [Change data capture](/concepts/capture/#one-owner). * A broker source consumes on every replica. The broker spreads the messages, and the inbox unique index rejects a repeat. * Stop every old replica before a migration that the upgrade notes mark as breaking. See [Upgrade QueueBox](/how-to/upgrade/). # Monitoring > What to scrape, what to alert on, and which metric answers which question about a running QueueBox. This page tells you what to watch on a running QueueBox and when to page someone. For every metric name, type and label, see [Metrics](/reference/metrics/). ## Scrape the metrics [Section titled “Scrape the metrics”](#scrape-the-metrics) QueueBox serves Prometheus metrics at `GET /metrics`. The endpoint is on the management port when `server.managementPort` is set, and on the data port otherwise. Scrape every replica, because each replica reports its own counters. ```yaml scrape_configs: - job_name: queuebox metrics_path: /metrics static_configs: - targets: ["queuebox-0.internal:9090", "queuebox-1.internal:9090"] ``` QueueBox binds no JVM metrics. No `jvm_`, `process_` or `system_` family appears in the scrape. Use another exporter for heap, threads and garbage collection. ## Health endpoints [Section titled “Health endpoints”](#health-endpoints) | Endpoint | Answers | Use | | --------------- | --------------------------------------------- | --------------------------------- | | `/health/live` | 200 while the process runs | Liveness probe | | `/health/ready` | 200 when every component is up, 503 otherwise | Readiness probe and a basic alert | | `/health` | The same as `/health/ready` | Compatibility | The readiness body names each component and its status: `database`, `outbox-poller`, `inbox-relay`, each broker source, `retention-service` and `outbox-capture`. `outbox-capture` is advisory. Its fault shows in the body but never makes the answer 503, because delivery continues without capture. Read the body to see a capture fault. See [Deploy QueueBox](/operations/deploy/#health-probes). ## What to alert on [Section titled “What to alert on”](#what-to-alert-on) Each rule below names the metric, a starting threshold and the runbook scenario that answers it. Adjust the thresholds to your own delivery targets. | Alert | Condition | Runbook | | ------------------------ | ----------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------- | | Outbox delivery is late | `queuebox_outbox_oldest_pending_age_seconds > 300` for 5 minutes | [Scenario 3](/operations/runbook/#scenario-3-the-pending-gauge-grows) | | Inbox is late | `queuebox_inbox_oldest_pending_age_seconds > 300` for 5 minutes | [Scenario 7](/operations/runbook/#scenario-7-the-inbox-backlog-grows) | | Messages go dead | `increase(queuebox_outbox_messages_total{status="dead"}[15m]) > 0` | [Scenario 1](/operations/runbook/#scenario-1-inspect-dead-lettered-messages) | | A destination fails | `rate(queuebox_outbox_destination_messages_total{outcome="failure"}[5m])` above 10% of all outcomes of that destination | [Scenario 5](/operations/runbook/#scenario-5-a-destination-is-slow) | | Claims are lost | `increase(queuebox_claims_lost_total[15m]) > 0` | [Scenario 6](/operations/runbook/#scenario-6-claims-are-lost) | | The pool is starved | `hikaricp_connections_pending > 0` for 5 minutes, or `increase(hikaricp_connections_timeout_total[5m]) > 0` | [Scenario 4](/operations/runbook/#scenario-4-size-the-pool-and-the-batch) | | The inbox cannot store | `increase(queuebox_inbox_rejections_total{reason="storage_failed"}[5m]) > 0` | Check the database. | | The relay fails | `increase(queuebox_inbox_relay_errors_total[15m]) > 0` | [Scenario 7](/operations/runbook/#scenario-7-the-inbox-backlog-grows) | | An instance is not ready | `/health/ready` answers 503 for 2 minutes | Read the component that is `down`. | ### Why the age and not the count [Section titled “Why the age and not the count”](#why-the-age-and-not-the-count) Alert on `queuebox_outbox_oldest_pending_age_seconds`, not on `queuebox_outbox_messages_pending`. A large count can be a busy but healthy poller. A small count can be a stopped poller. The age of the oldest pending row separates the two: it grows only when delivery falls behind. The age gauges refresh on the poll cycle, at most once per `outbox.pendingGaugeIntervalMs` or `inbox.relay.pendingGaugeIntervalMs` (default 5000 ms). A gauge can therefore lag by up to that interval. The gauge runs no query on a scrape. Note The inbox relay refreshes `queuebox_inbox_oldest_pending_age_seconds`. With `inbox.relay.enabled: false`, the gauge stays at zero. Measure a pull-only inbox with SQL, or with the metrics of your pull workers. ### A retry is not a failure yet [Section titled “A retry is not a failure yet”](#a-retry-is-not-a-failure-yet) `queuebox_outbox_messages_total{status="failed"}` counts scheduled retries. A retry is normal when a destination has a short fault. Alert on `dead` messages and on the oldest-pending age, which show the effect on delivery. Use `failed` and `queuebox_http_publish_responses_total{status_class="5xx"}` on a dashboard to see a fault early. ### A duplicate is always visible [Section titled “A duplicate is always visible”](#a-duplicate-is-always-visible) `queuebox_claims_lost_total{component="outbox"}` counts deliveries that a destination received twice because a claim expired during the publish. A value above zero is not data loss. It says that work outlives its lease. See [Claims and leases](/concepts/claims-and-leases/#lost-claims-and-duplicates). ## Dashboards [Section titled “Dashboards”](#dashboards) A delivery dashboard answers four questions: | Question | Metrics | | ---------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | | Does delivery keep up? | `queuebox_outbox_oldest_pending_age_seconds`, `queuebox_outbox_messages_pending`, `queuebox_outbox_queue_depth` per destination | | Do destinations accept? | `queuebox_outbox_destination_messages_total` by `destination` and `outcome`, `queuebox_http_publish_responses_total` by `status_class` | | Is a destination slow? | `queuebox_outbox_publish_duration_seconds` by `destination_type`, and its `_max` gauge | | Does the inbox take traffic? | `queuebox_inbox_messages_total` by `status`, `queuebox_inbox_rejections_total` by `reason`, `queuebox_inbox_filtered_total` by `source` | Add `hikaricp_connections_active` and `hikaricp_connections_pending` for the pool, and `queuebox_transform_failures_total` when you use transforms. ## Retention [Section titled “Retention”](#retention) When retention is on, watch that the cleanup runs: * `queuebox_cleanup_last_run_timestamp` per `table` must advance once per `cleanupInterval`. Alert when `time() - queuebox_cleanup_last_run_timestamp` exceeds twice the interval. * `queuebox_cleanup_messages_deleted_total` shows how many rows each run removes. A cleanup that stops lets `sent` and `dead` rows grow. The claim then scans a larger table. ## Capture [Section titled “Capture”](#capture) When change data capture is on, watch two more things: * The `outbox-capture` component of `/health/ready`. * On PostgreSQL, the replication slot. A slot that capture does not read holds write-ahead log and fills the database disk. ```sql SELECT slot_name, active, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained FROM pg_replication_slots; ``` See [Change data capture](/concepts/capture/). # Runbook > Diagnose and fix dead messages, a growing backlog, pool pressure, slow destinations, lost claims and a stalled inbox. This runbook gives the SQL and the commands for the common operational scenarios. Each scenario starts from a symptom and ends with actions in order. ## How to read the SQL [Section titled “How to read the SQL”](#how-to-read-the-sql) The SQL is for the shipped PostgreSQL schema. A test runs every `sql` block on this page against that schema, so a statement that drifts from the schema fails the build. * Each statement ends with a semicolon at the end of a line. * A placeholder starts with a colon, for example `:message_id`. Replace it with a real value before you run the statement. * The placeholders are `:message_id`, `:topic`, `:state`, `:destination` and `:limit`. * A `bash` or `yaml` block is a command or a configuration, not SQL. On SQL Server, write `SELECT TOP (n)` in place of `LIMIT n`, and `N'text'` for a string literal. The examples read the metrics on port 8080. Use the management port when `server.managementPort` is set. *** ## Scenario 1: Inspect dead-lettered messages [Section titled “Scenario 1: Inspect dead-lettered messages”](#scenario-1-inspect-dead-lettered-messages) **Symptom:** `queuebox_outbox_messages_total{status="dead"}` rises. A message reaches the state `dead` when its `attempt` count reaches the `max_attempts` of its own row. QueueBox writes `outbox.maxAttempts` into that column for every row that it creates, and your application can set a different value per row. The column `last_error` holds the redacted reason for the last failure. Count the dead messages per topic: ```sql SELECT topic, count(*) AS dead_count FROM outbox WHERE state = 'dead' GROUP BY topic ORDER BY dead_count DESC; ``` List the most recent dead messages with the reason: ```sql SELECT id, topic, key, attempt, max_attempts, updated_at, last_error FROM outbox WHERE state = 'dead' ORDER BY updated_at DESC LIMIT :limit; ``` Read one message in full: ```sql SELECT id, topic, key, payload, headers, attempt, scheduled_at, created_at, updated_at, claimed_at, last_error FROM outbox WHERE id = :message_id; ``` Group the dead messages by the first part of the failure reason: ```sql SELECT left(last_error, 60) AS reason, count(*) AS dead_count FROM outbox WHERE state = 'dead' GROUP BY left(last_error, 60) ORDER BY dead_count DESC; ``` The inbox has a `dead` state too. List the dead inbox rows: ```sql SELECT id, source, idempotency_key, event_type, created_at FROM inbox WHERE state = 'dead' ORDER BY created_at DESC LIMIT :limit; ``` Read the dead count over time from the metrics: ```bash curl -s http://localhost:8080/metrics | grep queuebox_outbox_messages_total ``` A reason of `No route matches topic` means that no route pattern matches the topic. Add the route, then replay the messages. *** ## Scenario 2: Replay a dead-lettered message [Section titled “Scenario 2: Replay a dead-lettered message”](#scenario-2-replay-a-dead-lettered-message) [Dead letters](/operations/dead-letters/) holds the full SQL procedure. [Replay dead letters](/how-to/replay-dead-letters/) replays through the admin API with filters. This section gives the short SQL form. Correct the cause of the failure first. A replay against a destination that is still broken produces a second dead message. Replay one message: ```sql UPDATE outbox SET state = 'pending', attempt = 0, scheduled_at = CURRENT_TIMESTAMP, claimed_at = NULL, last_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = :message_id AND state = 'dead'; ``` Replay every dead message of one topic: ```sql UPDATE outbox SET state = 'pending', attempt = 0, scheduled_at = CURRENT_TIMESTAMP, claimed_at = NULL, last_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE state = 'dead' AND topic = :topic; ``` Confirm the result: ```sql SELECT id, state, attempt, scheduled_at FROM outbox WHERE id = :message_id; ``` The message is delivered when its state becomes `sent`. *** ## Scenario 3: The pending gauge grows [Section titled “Scenario 3: The pending gauge grows”](#scenario-3-the-pending-gauge-grows) **Symptom:** `queuebox_outbox_oldest_pending_age_seconds` or `queuebox_outbox_messages_pending` grows. A growing backlog means that the poller delivers slower than the producers write. The oldest-pending age is the better signal. A count of pending rows cannot separate a busy poller from a stopped one. Read the gauges: ```bash curl -s http://localhost:8080/metrics | grep -E 'queuebox_outbox_messages_pending|queuebox_outbox_oldest_pending_age_seconds' ``` Measure the backlog and its age: ```sql SELECT count(*) AS pending_count, min(created_at) AS oldest_created_at, max(created_at) AS newest_created_at FROM outbox WHERE state = 'pending'; ``` Find the oldest rows that are due now: ```sql SELECT id, topic, key, created_at, scheduled_at, attempt FROM outbox WHERE state = 'pending' AND scheduled_at <= CURRENT_TIMESTAMP ORDER BY scheduled_at ASC LIMIT :limit; ``` Separate a real backlog from a retry backlog. A large `attempt` value means that the destination rejects the messages: ```sql SELECT attempt, count(*) AS pending_count FROM outbox WHERE state = 'pending' GROUP BY attempt ORDER BY attempt ASC; ``` Find the keys that hold back other rows. A row of a key waits while an earlier row of the same key is `pending` or `processing`, so one retrying row can stall its key: ```sql SELECT key, count(*) AS waiting_rows, min(created_at) AS oldest_created_at FROM outbox WHERE state = 'pending' AND key IS NOT NULL AND key <> '' GROUP BY key ORDER BY waiting_rows DESC LIMIT :limit; ``` Check for rows that stay in the state `processing`. The reclaim step returns such a row to `pending` after its lease of `outbox.claimTimeoutMs` expires: ```sql SELECT count(*) AS stuck_count, min(claimed_at) AS oldest_claim FROM outbox WHERE state = 'processing'; ``` Actions, in order: 1. Confirm that the destination is healthy. Use scenario 5. 2. Raise `outbox.concurrency` if the destination accepts more parallel requests. 3. Raise `outbox.batchSize` if each poll cycle returns a full batch. 4. Lower `outbox.pollIntervalMs` if the batch is not full and the backlog still grows. 5. Raise `database.poolSize` if the pool limits the poller. Use scenario 4. A backlog on one key does not shrink with more concurrency. One row of a key is in flight at a time. See [Ordering](/concepts/ordering/#throughput-of-one-key). *** ## Scenario 4: Size the pool and the batch [Section titled “Scenario 4: Size the pool and the batch”](#scenario-4-size-the-pool-and-the-batch) **Symptom:** `hikaricp_connections_pending` is above zero, or claims are slow. Four settings control the throughput of the outbox poller: | Setting | Default | Effect | | ----------------------- | ------- | ---------------------------------------------------------------- | | `database.poolSize` | 10 | The maximum number of open database connections. | | `outbox.batchSize` | 100 | The number of messages that one poll cycle claims. | | `outbox.concurrency` | 8 | The number of messages that QueueBox publishes at the same time. | | `outbox.pollIntervalMs` | 100 | The wait between two poll cycles. | Rules: * Keep `database.poolSize` larger than `outbox.concurrency`. The poller, the relay, the retention job and the HTTP routes all take a connection. * Keep `outbox.batchSize` larger than `outbox.concurrency`. A batch smaller than the concurrency leaves publisher slots idle. * The destination is the usual limit, not the database. Raise the concurrency first. Set the values in the YAML file: ```yaml # fragment database: poolSize: 20 outbox: pollIntervalMs: 100 batchSize: 200 concurrency: 16 claimTimeoutMs: 300000 ``` Or set them through the environment: ```bash export QUEUEBOX_DATABASE_POOLSIZE=20 export QUEUEBOX_OUTBOX_BATCHSIZE=200 export QUEUEBOX_OUTBOX_CONCURRENCY=16 export QUEUEBOX_OUTBOX_POLLINTERVALMS=100 ``` Compare the open connections with `poolSize` times the number of replicas: ```sql SELECT count(*) AS open_connections FROM pg_stat_activity WHERE datname = current_database(); ``` Check the server limit: ```sql SHOW max_connections; ``` Read the pool metrics and the processing time together: ```bash curl -s http://localhost:8080/metrics | grep -E 'hikaricp_connections_(active|pending|timeout_total)|queuebox_outbox_processing_duration_seconds' ``` *** ## Scenario 5: A destination is slow [Section titled “Scenario 5: A destination is slow”](#scenario-5-a-destination-is-slow) **Symptom:** `queuebox_outbox_publish_duration_seconds` rises, and then the pending backlog rises. Read the publish duration per destination type: ```bash curl -s http://localhost:8080/metrics | grep queuebox_outbox_publish_duration_seconds ``` Read the failures per destination and the HTTP status classes: ```bash curl -s http://localhost:8080/metrics | grep -E 'queuebox_outbox_destination_messages_total|queuebox_http_publish_responses_total' ``` Read the readiness endpoint. It reports each component: ```bash curl -s http://localhost:8080/health/ready ``` Find the topics that retry. A retry is the first signal of a slow or failing destination: ```sql SELECT topic, count(*) AS retry_count, max(attempt) AS worst_attempt FROM outbox WHERE state = 'pending' AND attempt > 0 GROUP BY topic ORDER BY retry_count DESC; ``` Read the last error text for the affected topic: ```sql SELECT id, attempt, updated_at, last_error FROM outbox WHERE topic = :topic AND last_error IS NOT NULL ORDER BY updated_at DESC LIMIT :limit; ``` Measure the time between the creation and the last update of the delivered messages. A large value means a slow destination or a long retry chain: ```sql SELECT topic, count(*) AS sent_count, avg(extract(epoch FROM (updated_at - created_at))) AS avg_seconds FROM outbox WHERE state = 'sent' GROUP BY topic ORDER BY avg_seconds DESC; ``` Actions, in order: 1. Test the destination directly. Compare its latency with the configured `timeoutMs` of the destination. 2. Raise the destination `timeoutMs` if the destination is slow but correct. 3. Lower `outbox.concurrency` if the destination rejects requests under load. 4. Raise `outbox.retryBaseDelayMs` to give the destination more time between attempts. 5. Inspect the dead messages with scenario 1 after the destination recovers. *** ## Scenario 6: Claims are lost [Section titled “Scenario 6: Claims are lost”](#scenario-6-claims-are-lost) **Symptom:** `queuebox_claims_lost_total` rises. A lost claim means that a worker held a message past its lease, and another replica took it. For the outbox, the destination received a duplicate. For the inbox, the relay rolled its outbox insert back, so no duplicate exists. See [Claims and leases](/concepts/claims-and-leases/#lost-claims-and-duplicates). ```bash curl -s http://localhost:8080/metrics | grep queuebox_claims_lost_total ``` Look for rows that other replicas reclaimed: ```bash curl -s http://localhost:8080/metrics | grep queuebox_outbox_messages_reclaimed_total ``` Actions, in order: 1. Read the `component` label. `outbox` points at `outbox.claimTimeoutMs`, and `inbox` points at `inbox.relay.claimTimeoutMs`. 2. Compare the slowest publish in `queuebox_outbox_publish_duration_seconds_max` with the claim timeout. 3. Check the database latency and the pool. A renewal that cannot reach the database cannot extend the lease. 4. Check for long pauses of the process, for example CPU throttling of the container. 5. Raise the claim timeout above the slowest publish or forward. *** ## Scenario 7: The inbox backlog grows [Section titled “Scenario 7: The inbox backlog grows”](#scenario-7-the-inbox-backlog-grows) **Symptom:** `queuebox_inbox_oldest_pending_age_seconds` grows. The gauge covers push and pull rows. A push row waits for the relay. A pull row waits for your worker. The relay refreshes the gauge, so the gauge stays at zero when `inbox.relay.enabled` is false. Use the SQL below in that case. Find which source and which mode hold the backlog: ```sql SELECT source, consumption, count(*) AS pending_count, min(created_at) AS oldest_created_at FROM inbox WHERE state = 'pending' GROUP BY source, consumption ORDER BY oldest_created_at ASC; ``` Find the aggregates that hold back other rows: ```sql SELECT source, aggregate_id, count(*) AS waiting_rows FROM inbox WHERE state = 'pending' AND aggregate_id IS NOT NULL GROUP BY source, aggregate_id ORDER BY waiting_rows DESC LIMIT :limit; ``` Actions, in order: 1. For a `pull` backlog, check the workers of that source. QueueBox does not move pull rows. 2. For a `push` backlog, read the `inbox-relay` component of `/health/ready` and the counter `queuebox_inbox_relay_errors_total`. 3. Confirm that `inbox.relay.enabled` is true. 4. Raise `inbox.relay.batchSize` or lower `inbox.relay.pollIntervalMs` if the relay keeps up but lags. Note On SQL Server, the pull claim serializes per source. More workers on one busy source do not raise its throughput. See [Ordering](/concepts/ordering/#the-sql-server-pull-claim-serializes-per-source). # Security > Terminate TLS in front of QueueBox, keep secrets in files, restrict outbound calls, and lock down the admin routes. This page covers the transport, the secrets, the outbound calls and the exposed endpoints. Read it before you put QueueBox on a network that you do not control. ## Transport security [Section titled “Transport security”](#transport-security) QueueBox listens on plain HTTP. It does not terminate TLS. Put a reverse proxy or an ingress in front of it, and terminate TLS there. Certificate rotation then stays out of the application. Never expose the QueueBox port directly to the internet. Publish only the inbox path. `/metrics`, `/health` and `/admin` stay inside your network. Set `server.managementPort` to move them to a separate port, so that an ingress rule on the data port cannot publish them by mistake. ### Kubernetes ingress example [Section titled “Kubernetes ingress example”](#kubernetes-ingress-example) The example terminates TLS at the ingress and sends plain HTTP to the service inside the cluster. It limits the request body to the same value as `inbox.maxBodyBytes`. ```yaml apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: queuebox annotations: cert-manager.io/cluster-issuer: letsencrypt-prod nginx.ingress.kubernetes.io/proxy-body-size: 1m nginx.ingress.kubernetes.io/ssl-redirect: "true" spec: ingressClassName: nginx tls: - hosts: - webhooks.example.com secretName: queuebox-tls rules: - host: webhooks.example.com http: paths: - path: /inbox pathType: Prefix backend: service: name: queuebox port: number: 8080 --- apiVersion: v1 kind: Service metadata: name: queuebox spec: selector: app: queuebox ports: - port: 8080 targetPort: 8080 ``` ### Nginx example [Section titled “Nginx example”](#nginx-example) ```nginx server { listen 443 ssl http2; server_name webhooks.example.com; ssl_certificate /etc/ssl/certs/queuebox.crt; ssl_certificate_key /etc/ssl/private/queuebox.key; ssl_protocols TLSv1.2 TLSv1.3; client_max_body_size 1m; location /inbox/ { proxy_pass http://queuebox:8080; proxy_set_header Host $host; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } } ``` ## Inbound authentication [Section titled “Inbound authentication”](#inbound-authentication) An HTTP source can require a bearer token, an API key or an HMAC signature. A request that fails the check gets 401, and QueueBox stores nothing. See [Authenticate requests](/how-to/authenticate-requests/). Authenticate every source that the internet can reach. A source without `auth` accepts any caller that reaches the path. ## Outbound calls [Section titled “Outbound calls”](#outbound-calls) QueueBox calls an HTTP destination with the scheme that `destinations..baseUrl` names. Use `https://` for every destination that leaves your network. QueueBox refuses a base URL that is not an absolute HTTP or HTTPS URL. QueueBox also refuses: * a destination URL that carries a user name or a password, * a destination path that carries a `.` or a `..` segment. QueueBox never follows a redirect for a destination. A 3xx answer fails the publish, and the retry or the dead-letter path runs. Without this rule, a public destination could redirect QueueBox to a metadata address with the destination credentials attached. ### Block private addresses [Section titled “Block private addresses”](#block-private-addresses) Set `http.blockPrivateAddresses: true` when the destination configuration comes from a layer that you trust less. QueueBox then refuses a destination that resolves to a loopback address, a link-local address or a private range. The same check covers the OAuth2 `tokenUrl`, which carries the client secret in its request body. ```yaml # fragment http: blockPrivateAddresses: true ``` Caution The address check runs once, at start. The publisher resolves the host again on every request. A DNS name whose record changes to a private address after the start still passes. Where the destination configuration is not trusted, add an egress policy at the network. Only the network can enforce the rule at the time of each request. ## Secrets [Section titled “Secrets”](#secrets) Every credential field accepts a `file:` reference. QueueBox reads the file once, at start, and removes the trailing newline. ```yaml # fragment database: password: file:/run/secrets/queuebox-db-password sources: stripe: type: http path: /stripe idempotencyKeyPath: $.id eventTypePath: $.type auth: type: hmac secret: file:/run/secrets/stripe-webhook-secret ``` A `file:` reference works only on a field that is a credential and nothing else. `database.url`, the RabbitMQ destination `url` and the RabbitMQ source `connectionUrl` carry a password inside a URL. Supply those through an environment variable, for example `QUEUEBOX_DATABASE_URL`. On one of those fields, QueueBox reads a `file:` string literally, and `database.url` then fails validation at start. ### The Kubernetes secret pattern [Section titled “The Kubernetes secret pattern”](#the-kubernetes-secret-pattern) Mount the secret as a file, not as an environment variable. Another process cannot read a file from `/proc//environ`. You can also rotate a file without a change to the pod template. ```yaml apiVersion: v1 kind: Secret metadata: name: queuebox-secrets type: Opaque stringData: db-password: the-real-password stripe-webhook-secret: whsec_the_real_secret --- apiVersion: apps/v1 kind: Deployment metadata: name: queuebox spec: selector: matchLabels: app: queuebox template: metadata: labels: app: queuebox spec: containers: - name: queuebox image: ghcr.io/alternayte/queuebox:1.2.3 env: - name: QUEUEBOX_DATABASE_PASSWORD value: file:/run/secrets/db-password volumeMounts: - name: secrets mountPath: /run/secrets readOnly: true volumes: - name: secrets secret: secretName: queuebox-secrets ``` An external secret manager works the same way. Have it write a file, and point the configuration at the path. ### Secrets in logs [Section titled “Secrets in logs”](#secrets-in-logs) A credential field prints as a mask. A log line, an exception message or a crash dump that prints a configuration object therefore shows no secret. * A value that is not empty prints as `Secret(***)`. * An empty value prints as `Secret(empty)`, which tells an operator that the credential is absent. * A JDBC URL or an AMQP URI prints with its password masked. The publisher also redacts the reason for a failed delivery before it stores it in `last_error`. It masks the value of every known secret-bearing key and truncates the text. ## The admin routes [Section titled “The admin routes”](#the-admin-routes) The admin routes change data and run code that the caller supplies: * `POST /admin/transform/test` evaluates a JSONata expression on the host that processes your messages. * `POST /admin/replay` sends delivered and dead outbox rows again. See [Replay dead letters](/how-to/replay-dead-letters/). QueueBox protects them as follows: * The routes do not exist until you set `admin.enabled: true`. * They need authentication. QueueBox refuses to start with `admin.enabled: true` and no `admin.auth`, unless you set `admin.insecure: true`. Never set `admin.insecure` in production. * QueueBox clamps the caller’s transform timeout to `admin.maxTransformTimeoutMs` (default 1000 ms). * QueueBox rejects a request body larger than `admin.maxPayloadBytes` (default 65536 bytes) with 413. ```yaml # fragment admin: enabled: true auth: type: bearer token: file:/run/secrets/queuebox-admin-token ``` Never publish `/admin` through the ingress. ## Request limits [Section titled “Request limits”](#request-limits) | Setting | Default | Purpose | | -------------------------------------------- | ------- | ---------------------------------------------------------------------------------------------- | | `inbox.maxBodyBytes` | 1048576 | The largest accepted request body. A larger body gets 413. | | `sources..rateLimit.requestsPerMinute` | Not set | The request rate that one source accepts. A caller over the limit gets 429 with `Retry-After`. | | `http.maxErrorBodyBytes` | 2048 | The largest error body that a failed delivery keeps. | The rate limit uses one bucket per source. It protects the database from one busy source. It does not separate one caller from another, so keep the per-client limit at the ingress. # Configuration > Every section and key of queuebox.yml, with its type, its default and its meaning. This page lists every key of `queuebox.yml`, with its type, its default and its meaning. The [environment variables page](/reference/environment-variables/) gives the `QUEUEBOX_` name of each key and the precedence of the sources. ## Where the configuration comes from [Section titled “Where the configuration comes from”](#where-the-configuration-comes-from) QueueBox reads its configuration from three places: 1. The `QUEUEBOX_` environment variables. 2. One YAML file. `QUEUEBOX_CONFIG_FILE` names the file. The default path is `/etc/queuebox/queuebox.yml`. 3. The `queuebox.yml` resource in the image. A variable wins over the file. The file replaces the packaged resource, so the file must hold a complete configuration. QueueBox reads the packaged resource only when no file exists and no `QUEUEBOX_` variable is set. Start from [`examples/queuebox.yml`](https://github.com/alternayte/queuebox/blob/main/examples/queuebox.yml). Do not edit `config/src/main/resources/queuebox.yml`, because a change there needs a new image. ## Value formats [Section titled “Value formats”](#value-formats) | Format | Where | Rule | | ------------------ | ------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Milliseconds | Every key that ends in `Ms` | A whole number of milliseconds. | | Retention duration | `retention.*.maxAge`, `retention.*.cleanupInterval` | A whole number and one suffix: `s`, `m`, `h` or `d`. For example `7d`. | | Secret | Every password, token, key and client secret | A string, or `file:` and a path. QueueBox reads the file once at load time and removes the trailing newline. A log line prints `Secret(***)`. | | Placeholder | Any value | `${NAME}` reads the environment variable or JVM system property `NAME`. `${NAME:-fallback}` gives a fallback value. | | Enumeration | `transform.onError`, `retention.*.policy`, `auth.signaturePayloadFormat` | Write the value in the exact letter case that the tables below show. `onError: fail` does not load. `signaturePayloadFormat` also accepts `body` and `timestamp-dot-body`. | ## How QueueBox selects the kind of an entry [Section titled “How QueueBox selects the kind of an entry”](#how-queuebox-selects-the-kind-of-an-entry) A destination, a source and an `auth` block each have several kinds. Write `type` on every entry, so a reader sees the kind. The loader does not check `type`. It selects the kind from the required keys that the entry sets. An entry with `type: rabbitmq` and the keys of an HTTP destination loads as an HTTP destination. | Block | Kind | Required keys that select the kind | | ----------------------------------- | ---------- | --------------------------------------- | | `destinations.` | `http` | `baseUrl` | | `destinations.` | `rabbitmq` | `url`, `exchange` | | `destinations.` | `kafka` | `bootstrapServers`, `topic` | | `destinations.` | `nats` | `servers`, `subject` | | `sources.` | `http` | `path`, `idempotencyKeyPath` | | `sources.` | `rabbitmq` | `queueName`, `connectionUrl` | | `sources.` | `kafka` | `bootstrapServers`, `topics`, `groupId` | | `sources.` | `nats` | `servers`, `stream`, `durable` | | `destinations..auth` | `oauth2` | `clientId`, `clientSecret`, `tokenUrl` | | `destinations..auth` | `basic` | `username`, `password` | | `destinations..auth` | `header` | `headerValue` | | `sources..auth`, `admin.auth` | `bearer` | `token` | | `sources..auth`, `admin.auth` | `api-key` | `key` | | `sources..auth`, `admin.auth` | `hmac` | `secret` | ## A complete example [Section titled “A complete example”](#a-complete-example) This file sets every section. It loads and validates as it is, when the placeholders have values. ```yaml server: httpPort: 8080 # The data port: the inbox routes managementPort: 9090 # Moves /health, /metrics and /admin to this port database: type: postgresql # postgresql or sqlserver url: jdbc:postgresql://db:5432/queuebox username: queuebox password: ${DB_PASSWORD} # Or file:/run/secrets/db-password poolSize: 10 connectionTimeoutMs: 30000 startupTimeoutMs: 60000 # How long the start waits for the database migrate: true # Apply the bundled Flyway migrations at start outbox: pollIntervalMs: 100 batchSize: 100 concurrency: 8 # Rows published at the same time retryBaseDelayMs: 1000 # Backoff base: 1000 ms * 2^attempt, at most 60 s maxAttempts: 5 # Written into max_attempts of each row QueueBox creates claimTimeoutMs: 300000 # A claim older than this returns to pending pendingGaugeIntervalMs: 5000 shutdownTimeoutMs: 30000 capture: mode: polling # polling, postgres-logical or sqlserver-cdc enabled: false inbox: basePath: /inbox # A source with path /stripe answers at /inbox/stripe maxBodyBytes: 1048576 # A larger request body gets 413 relay: enabled: true # Copies push rows into the outbox pollIntervalMs: 100 batchSize: 100 claimTimeoutMs: 300000 pendingGaugeIntervalMs: 5000 maxAttempts: 10 # Relayed rows only. Default: outbox.maxAttempts http: maxErrorBodyBytes: 2048 # Bytes of a failed response kept in last_error blockPrivateAddresses: false # true refuses a destination on a private address admin: enabled: true # Registers /admin/transform/test and /admin/replay auth: type: bearer token: ${ADMIN_TOKEN} maxTransformTimeoutMs: 1000 maxPayloadBytes: 65536 destinations: orders-api: type: http baseUrl: https://orders.example.com path: /events timeoutMs: 30000 headers: X-Team: orders # A static header on every request auth: type: oauth2 clientId: queuebox clientSecret: ${ORDERS_CLIENT_SECRET} tokenUrl: https://auth.example.com/oauth/token transform: expression: '{ "data": $, "topic": $topic }' onError: Fail # Fail, Skip or Dead, in this letter case events-exchange: type: rabbitmq url: amqp://rabbitmq:5672 exchange: "events.{{ aggregateType }}" # A template, or a literal name exchangeType: topic routingKeyTemplate: "{{ topic }}" deliveryMode: persistent events-topic: type: kafka bootstrapServers: kafka-1:9092,kafka-2:9092 topic: orders keyTemplate: "{{ key }}" timeoutMs: 30000 # At least 2000 events-subject: type: nats servers: nats://nats:4222 subject: "orders.{{ topic }}" jetStream: true routes: - topicPattern: "order.*" # First match wins, in list order destination: orders-api - topicPattern: "event.**" destination: events-exchange routingKeyTemplate: "{{ payload.region }}.{{ topic }}" routingKeyMissingFieldDefault: unknown sources: stripe: type: http path: /stripe # POST /inbox/stripe idempotencyKeyPath: $.id eventTypePath: $.type aggregateIdPath: $.data.object.customer topic: "stripe.{{ eventType }}" auth: type: hmac secret: ${STRIPE_WEBHOOK_SECRET} headerName: Stripe-Signature signaturePrefix: "v1=" rateLimit: requestsPerMinute: 600 orders: type: http path: /orders idempotencyKeyPath: $.id consumption: pull # A pull client claims the row. No topic or route needed orders-queue: type: rabbitmq queueName: incoming-orders connectionUrl: amqp://rabbitmq:5672 idempotencyKeyPath: $.messageId aggregateIdPath: $.orderId topic: "{{ source }}" filter: exclude: - header: x-test exists: true orders-log: type: kafka bootstrapServers: kafka-1:9092 topics: [orders] groupId: queuebox-orders idempotencyKeyPath: $.id eventTypePath: $.type topic: "{{ eventType }}" orders-stream: type: nats servers: nats://nats:4222 stream: ORDERS durable: queuebox-orders filterSubject: "orders.>" idempotencyKeyPath: $.id retention: enabled: true outbox: policy: AGE # AGE, COUNT or DISABLED, in upper case maxAge: 7d cleanupInterval: 1h batchSize: 1000 inbox: policy: AGE # AGE or DISABLED maxAge: 30d cleanupInterval: 6h batchSize: 1000 ``` ## server [Section titled “server”](#server) | Key | Type | Default | Meaning | | ----------------------- | ------- | ------- | ----------------------------------------------------------------------------------------------------------------------------------------- | | `server.httpPort` | integer | `8080` | The port of the inbox routes. It must be from 1 to 65535. | | `server.managementPort` | integer | none | The port of `/health`, `/metrics` and `/admin`. When it is set, the data port answers 404 on those paths. It must differ from `httpPort`. | ## database [Section titled “database”](#database) | Key | Type | Default | Meaning | | --------------------------------- | ------------ | ----------------- | --------------------------------------------------------------------------------------------------------------- | | `database.type` | string | `postgresql` | `postgresql` or `sqlserver`. | | `database.url` | string | required | The JDBC URL. It must start with `jdbc:postgresql://` or `jdbc:sqlserver://`, to match `type`. | | `database.username` | string | required | The database user. | | `database.password` | secret | required | The database password. | | `database.poolSize` | integer | `10` | The size of the HikariCP connection pool. It must be above 0. | | `database.connectionTimeoutMs` | milliseconds | `30000` | The longest wait for a connection from the pool. | | `database.startupTimeoutMs` | milliseconds | `60000` | How long the start retries the database before the process exits. | | `database.migrate` | boolean | `true` | Apply the bundled Flyway migrations at start. Set it to `false` when you rename a table or a column. | | `database.outboxTableName` | string | `outbox` | The name of the outbox table. It must be a SQL identifier: a letter or `_` first, then letters, digits and `_`. | | `database.inboxTableName` | string | `inbox` | The name of the inbox table. The same identifier rule applies. | | `database.columnMapping.outbox.*` | string | the default names | The outbox column names. See the next table. | | `database.columnMapping.inbox.*` | string | the default names | The inbox column names. See the table after it. | QueueBox refuses to start when a table or a column has a custom name and `database.migrate` is `true`. The bundled migrations create the default names only. Apply your own schema first. [Use custom tables](/how-to/use-custom-tables/) gives the steps. ### database.columnMapping.outbox [Section titled “database.columnMapping.outbox”](#databasecolumnmappingoutbox) Each key names one column of the outbox table. A key that you omit keeps its default. Every value must be a SQL identifier. The [outbox table](/reference/outbox-table/) page describes each column. | Key | Default column | | ---------------- | ------------------ | | `id` | `id` | | `topic` | `topic` | | `key` | `key` | | `aggregateType` | `aggregate_type` | | `payload` | `payload` | | `headers` | `headers` | | `state` | `state` | | `attempt` | `attempt` | | `maxAttempts` | `max_attempts` | | `scheduledAt` | `scheduled_at` | | `createdAt` | `created_at` | | `updatedAt` | `updated_at` | | `claimedAt` | `claimed_at` | | `claimToken` | `claim_token` | | `leaseExpiresAt` | `lease_expires_at` | | `lastError` | `last_error` | | `sequence` | `sequence` | ### database.columnMapping.inbox [Section titled “database.columnMapping.inbox”](#databasecolumnmappinginbox) Each key names one column of the inbox table. A key that you omit keeps its default. The [inbox table](/reference/inbox-table/) page describes each column. | Key | Default column | | ---------------- | ------------------ | | `id` | `id` | | `source` | `source` | | `idempotencyKey` | `idempotency_key` | | `aggregateId` | `aggregate_id` | | `eventType` | `event_type` | | `payload` | `payload` | | `state` | `state` | | `createdAt` | `created_at` | | `processedAt` | `processed_at` | | `claimedAt` | `claimed_at` | | `claimToken` | `claim_token` | | `leaseExpiresAt` | `lease_expires_at` | | `correlationId` | `correlation_id` | | `consumption` | `consumption` | | `scheduledAt` | `scheduled_at` | | `attempt` | `attempt` | | `lastError` | `last_error` | | `headers` | `headers` | At start, QueueBox reads the columns of the inbox table. It stops when the column that `headers` names does not exist. The error message holds the `ALTER TABLE` statement that adds the column. ## outbox [Section titled “outbox”](#outbox) | Key | Type | Default | Meaning | | ------------------------------- | ------------ | -------- | ------------------------------------------------------------------------------------------------------------ | | `outbox.pollIntervalMs` | milliseconds | `100` | The wait between two claim cycles of the poller. It must be above 0. | | `outbox.batchSize` | integer | `100` | The largest number of rows that one claim takes. It must be above 0. | | `outbox.concurrency` | integer | `8` | The largest number of rows that the poller publishes at the same time. It must be above 0. | | `outbox.retryBaseDelayMs` | milliseconds | `1000` | The base of the retry backoff. See the note below. | | `outbox.maxAttempts` | integer | `5` | The dead-letter ceiling that QueueBox writes into `max_attempts` of each row it creates. It must be above 0. | | `outbox.claimTimeoutMs` | milliseconds | `300000` | The visibility timeout. A claim older than this returns to pending. Set it above the slowest publish. | | `outbox.pendingGaugeIntervalMs` | milliseconds | `5000` | The shortest interval between two queries for the pending gauges. | | `outbox.shutdownTimeoutMs` | milliseconds | `30000` | The longest wait for the publishes in flight at shutdown. | | `outbox.capture.*` | block | polling | Change data capture. See the next section. | The retry delay after a failed delivery is `retryBaseDelayMs × 2^attempt`, plus a random 0 to 25 percent. The delay never passes 60 seconds. The poller reads `max_attempts` of the row, not `outbox.maxAttempts`. A row that the application inserts takes the column default of `5` unless the insert sets `max_attempts`. ### outbox.capture [Section titled “outbox.capture”](#outboxcapture) Capture wakes the poller when the database log shows an outbox insert. The poller still claims and publishes through SQL. [Capture](/concepts/capture/) explains the model and [Capture changes](/how-to/capture-changes/) gives the setup. | Key | Type | Default | Meaning | | ----------------------------------------- | ------------ | ----------------- | ------------------------------------------------------------------------------------------------------------------------ | | `outbox.capture.mode` | string | `polling` | `polling`, `postgres-logical` or `sqlserver-cdc`. `postgres-logical` needs PostgreSQL. `sqlserver-cdc` needs SQL Server. | | `outbox.capture.enabled` | boolean | `false` | Set it to `true` on the one replica that owns capture. `true` needs a mode other than `polling`. | | `outbox.capture.identity` | string | `queuebox` | The name of the capture owner. Lower case letters, digits and `_`, at most 63 characters, a letter first. | | `outbox.capture.stateDirectory` | string | empty | A durable directory for the capture offsets. Required when `enabled` is `true`. | | `outbox.capture.schema` | string | none | The schema of the outbox table. None means `public` on PostgreSQL and `dbo` on SQL Server. | | `outbox.capture.publication` | string | `queuebox_outbox` | PostgreSQL only. The publication of the outbox table. QueueBox never creates it. | | `outbox.capture.slot` | string | `queuebox_outbox` | PostgreSQL only. The replication slot. The connector creates it on the first start. | | `outbox.capture.reconciliationIntervalMs` | milliseconds | `1000` | The longest wait of the poller when no event arrives. It must be above 0. | The `identity`, `slot` and `publication` values follow the same rule: a lower case letter first, then lower case letters, digits and `_`, at most 63 characters. ### outbox.capture.connection [Section titled “outbox.capture.connection”](#outboxcaptureconnection) Capture reads the host, the port and the database from `database.url`. Each key below replaces one value for the capture connection only. | Key | Type | Default | Meaning | | -------------------------------------------------- | ------- | ------------------- | ------------------------------------------------------------------------ | | `outbox.capture.connection.hostname` | string | from `database.url` | The capture host. Required when `database.url` names more than one host. | | `outbox.capture.connection.port` | integer | from `database.url` | The capture port, from 1 to 65535. | | `outbox.capture.connection.database` | string | from `database.url` | The database name. | | `outbox.capture.connection.username` | string | `database.username` | The capture account. | | `outbox.capture.connection.password` | secret | `database.password` | The password of the capture account. | | `outbox.capture.connection.encrypt` | boolean | `true` | SQL Server only. Encrypt the capture connection. | | `outbox.capture.connection.trustServerCertificate` | boolean | `false` | SQL Server only. Accept the server certificate without a check. | ## inbox [Section titled “inbox”](#inbox) | Key | Type | Default | Meaning | | -------------------- | ------- | --------- | ---------------------------------------------------------------------------------------------------------------------------------------- | | `inbox.basePath` | string | `/inbox` | The prefix of every HTTP source path. A source with `path: /stripe` answers at `/inbox/stripe`. The source name is not part of the path. | | `inbox.maxBodyBytes` | integer | `1048576` | The largest request body in bytes, on every route of the data port. A larger body gets `413`. It must be above 0. | | `inbox.relay.*` | block | on | The relay. See the next section. | ### inbox.relay [Section titled “inbox.relay”](#inboxrelay) The relay copies each push row of the inbox into the outbox. A pull row stays in the inbox for a [pull client](/reference/pull-clients/). | Key | Type | Default | Meaning | | ------------------------------------ | ------------ | -------------------- | ------------------------------------------------------------------------------------------------------ | | `inbox.relay.enabled` | boolean | `true` | Run the relay. With `false`, a push row stays in the inbox. | | `inbox.relay.pollIntervalMs` | milliseconds | `100` | The wait between two relay cycles. | | `inbox.relay.batchSize` | integer | `100` | The largest number of rows that one relay claim takes. | | `inbox.relay.claimTimeoutMs` | milliseconds | `300000` | The visibility timeout of a relay claim. | | `inbox.relay.pendingGaugeIntervalMs` | milliseconds | `5000` | The shortest interval between two queries for the oldest pending inbox row. | | `inbox.relay.maxAttempts` | integer | `outbox.maxAttempts` | The value that the relay writes into `max_attempts` of each outbox row it creates. It must be above 0. | ## http [Section titled “http”](#http) These keys apply to every HTTP destination. | Key | Type | Default | Meaning | | ---------------------------- | ------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `http.maxErrorBodyBytes` | integer | `2048` | The largest part of a failed response body that QueueBox keeps. QueueBox removes secret values from the text before it reaches a log or `last_error`. | | `http.blockPrivateAddresses` | boolean | `false` | Refuse a destination `baseUrl` or OAuth2 `tokenUrl` whose host resolves to a loopback, link-local, site-local or unique-local address. The check runs at start. A host that does not resolve passes. | ## admin [Section titled “admin”](#admin) The admin routes evaluate a JSONata expression that the caller supplies, and they move outbox rows back to pending. The [HTTP API](/reference/http-api/#admin-routes) page describes the routes. | Key | Type | Default | Meaning | | ----------------------------- | ------------ | ------- | ------------------------------------------------------------------------------------------------------------------------------------- | | `admin.enabled` | boolean | `false` | Register `/admin/transform/test` and `/admin/replay`. With `false`, the routes do not exist. | | `admin.auth` | block | none | The credentials of the admin routes. It takes the same kinds as a [source `auth` block](#source-auth): `bearer`, `api-key` or `hmac`. | | `admin.insecure` | boolean | `false` | Allow the admin routes without `auth`. Use it for a local test only. | | `admin.maxTransformTimeoutMs` | milliseconds | `1000` | The upper bound of the `timeoutMs` that a caller of `/admin/transform/test` sends. QueueBox uses the smaller value. | | `admin.maxPayloadBytes` | integer | `65536` | The largest admin request body. A larger body gets `413`. | QueueBox refuses to start when `admin.enabled` is `true`, `admin.auth` is absent and `admin.insecure` is `false`. ## destinations [Section titled “destinations”](#destinations) `destinations` is a map. The key is the destination name, and a route refers to it. ### Keys of every destination [Section titled “Keys of every destination”](#keys-of-every-destination) | Key | Type | Default | Meaning | | ----------- | -------------- | ------- | --------------------------------------------------------------------------------------------------------------------------- | | `type` | string | none | `http`, `rabbitmq`, `kafka` or `nats`. See [how QueueBox selects the kind](#how-queuebox-selects-the-kind-of-an-entry). | | `headers` | map of strings | empty | Static headers on every message to this destination. A header of the outbox row wins over a static header of the same name. | | `transform` | block | none | A JSONata transform that runs before the publish. See [transform](#transform). | ### HTTP destination [Section titled “HTTP destination”](#http-destination) QueueBox sends a `POST` with the payload as the JSON body. A `2xx` status completes the delivery. | Key | Type | Default | Meaning | | ----------- | ------------ | -------- | ------------------------------------------------------------------------------------------ | | `baseUrl` | string | required | An absolute `http` or `https` URL with a host. It must not hold a user name or a password. | | `path` | string | `/` | The path after `baseUrl`. It must not hold a `.` or a `..` segment. | | `timeoutMs` | milliseconds | `30000` | The request timeout. | | `auth` | block | none | The credentials of the request. See the next section. | ### Destination auth [Section titled “Destination auth”](#destination-auth) | Kind | Key | Type | Default | Meaning | | -------- | -------------- | -------------- | --------------- | ---------------------------------------------------------- | | `oauth2` | `clientId` | string | required | The client identifier of the client credentials grant. | | `oauth2` | `clientSecret` | secret | required | The client secret. | | `oauth2` | `tokenUrl` | string | required | The token endpoint. The same URL rules as `baseUrl` apply. | | `oauth2` | `scope` | string | none | The `scope` parameter of the token request. | | `oauth2` | `extraParams` | map of strings | empty | Extra form parameters of the token request. | | `basic` | `username` | string | required | The user name of HTTP Basic authentication. | | `basic` | `password` | secret | required | The password. | | `header` | `headerName` | string | `Authorization` | The header that carries the credential. | | `header` | `headerValue` | secret | required | The value of that header. | The `oauth2` kind sends `Authorization: Bearer ` and caches the token. The `basic` kind sends `Authorization: Basic `. ### RabbitMQ destination [Section titled “RabbitMQ destination”](#rabbitmq-destination) QueueBox declares the exchange, publishes with the `mandatory` flag and waits for the broker confirm. An unroutable message fails and retries. | Key | Type | Default | Meaning | | -------------------- | ------ | ------------- | ---------------------------------------------------------------------------------------------------------------------------- | | `url` | string | required | The AMQP URI of the broker. | | `exchange` | string | required | The exchange name, or an [address template](#address-templates-and-address-columns). | | `exchangeFrom` | string | none | A row column that holds the exchange name: `aggregate_type`, `topic` or `key`. It wins over `exchange`. | | `exchangeType` | string | `topic` | The exchange type that QueueBox declares: `topic`, `direct` or `fanout`. | | `routingKeyTemplate` | string | `{{ topic }}` | The routing key when the matched route sets none. See [routing key templates](/reference/transforms/#routing-key-templates). | | `deliveryMode` | string | `persistent` | `persistent` or `transient`. A persistent message in a durable queue survives a broker restart. | ### Kafka destination [Section titled “Kafka destination”](#kafka-destination) The producer publishes with `acks=all` and idempotence. | Key | Type | Default | Meaning | | ------------------ | ------------ | ----------- | ---------------------------------------------------------------------------------------------------------------------- | | `bootstrapServers` | string | required | The brokers, separated by commas. | | `topic` | string | required | The Kafka topic, or an [address template](#address-templates-and-address-columns). | | `topicFrom` | string | none | A row column that holds the topic name: `aggregate_type`, `topic` or `key`. It wins over `topic`. | | `keyTemplate` | string | `{{ key }}` | The record key. `{{ key }}` and `{{ topic }}` render. An empty result sends no key. A route `routingKeyTemplate` wins. | | `timeoutMs` | milliseconds | `30000` | The whole publish budget, with the broker acknowledgement. It must be at least `2000`. | | `securityProtocol` | string | `PLAINTEXT` | `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT` or `SASL_SSL`. | | `saslMechanism` | string | none | For example `SCRAM-SHA-512`. Required by a `SASL_` protocol. | | `saslUsername` | string | none | Required by a `SASL_` protocol. | | `saslPassword` | secret | none | Required by a `SASL_` protocol. | ### NATS destination [Section titled “NATS destination”](#nats-destination) | Key | Type | Default | Meaning | | ------------- | ------------ | -------- | -------------------------------------------------------------------------------------------------------------------- | | `servers` | string | required | The server URLs, separated by commas. | | `subject` | string | required | The subject, or an [address template](#address-templates-and-address-columns). | | `subjectFrom` | string | none | A row column that holds the subject: `aggregate_type`, `topic` or `key`. It wins over `subject`. | | `jetStream` | boolean | `true` | Publish through JetStream and wait for its acknowledgement. `false` publishes on core NATS, with no acknowledgement. | | `timeoutMs` | milliseconds | `30000` | The publish timeout. It must be above 0. | | `username` | string | none | A user name. It needs `password`. | | `password` | secret | none | A password. It needs `username`. | | `token` | secret | none | A token. Set a token or a user name, not both. | ### Address templates and address columns [Section titled “Address templates and address columns”](#address-templates-and-address-columns) `exchange`, `topic` and `subject` accept a literal name or a template with `{{ ... }}` placeholders. | Placeholder | Value | | --------------------- | ----------------------------------------------- | | `{{ topic }}` | The `topic` column of the row. | | `{{ key }}` | The `key` column of the row. | | `{{ aggregateType }}` | The `aggregate_type` column of the row. | | `{{ payload.field }}` | A payload field. A dot separates nested fields. | | `{{ data.field }}` | The same as `payload.field`. | A placeholder with another name stops the start. The error names the field and the destination. A template that renders an empty name fails the publish of that row. `exchangeFrom`, `topicFrom` and `subjectFrom` name a column instead. The value of the column is the address, and QueueBox renders no template. The permitted names are `aggregate_type`, `topic` and `key`, with the underscore. Another name stops the start. ## routes [Section titled “routes”](#routes) `routes` is a list. QueueBox tests the patterns in list order, and the first match wins. A row that matches no route goes dead. | Key | Type | Default | Meaning | | ------------------------------- | ------ | ------------ | --------------------------------------------------------------------------------------------------------------------------------------------------- | | `topicPattern` | string | required | A glob on the whole topic. `*` matches one segment without a dot. `**` matches any text. Every other character is a literal. `***` stops the start. | | `destination` | string | required | The name of a destination. An unknown name stops the start. | | `routingKeyTemplate` | string | none | The RabbitMQ routing key or the Kafka record key for this route. It wins over the destination template. | | `routingKeyMissingFieldDefault` | string | empty string | The value of a placeholder that finds no value. | | `transform` | block | none | A JSONata transform that runs before the destination transform. See [transform](#transform). | ## sources [Section titled “sources”](#sources) `sources` is a map. The key is the source name. QueueBox stores the name in the `source` column of each inbox row. ### Keys of every source [Section titled “Keys of every source”](#keys-of-every-source) | Key | Type | Default | Meaning | | ----------------------------- | ------- | --------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------- | | `type` | string | none | `http`, `rabbitmq`, `kafka` or `nats`. See [how QueueBox selects the kind](#how-queuebox-selects-the-kind-of-an-entry). | | `consumption` | string | `push` | `push`: the relay copies the row into the outbox. `pull`: a pull client claims the row. QueueBox stores the value on the row at receipt. | | `topic` | string | `{{ eventType }}` for HTTP, `{{ source }}` for the others | The outbox topic that the relay writes. `{{ source }}` and `{{ eventType }}` render. It must not be blank on a push source. | | `idempotencyKeyPath` | string | required for HTTP, `$.id` for the others | The JSONPath of the idempotency key in the body. | | `aggregateIdPath` | string | none | The JSONPath of the aggregate identifier. The relay writes it into the outbox `key`. | | `eventTypePath` | string | none | The JSONPath of the event type. | | `transform` | block | none | A JSONata transform that runs before the store. See [transform](#transform). | | `filter` | block | none | A header filter. See [filter](#filter). | | `rateLimit.requestsPerMinute` | integer | none | The request limit of an HTTP source. A request over it gets `429`. QueueBox ignores the key on the other kinds. It must be above 0. | Every JSONPath must be definite. A path such as `$..orderId` stops the start. The relay marks a row dead when its topic renders empty. QueueBox therefore refuses to start when a push source uses `{{ eventType }}` and has no source of the event type. An HTTP source needs `eventTypePath`. The other kinds need `eventTypePath` or `eventTypeFromHeader: true`. ### HTTP source [Section titled “HTTP source”](#http-source) The source answers `POST `. The [HTTP API](/reference/http-api/#inbox-sources) page lists the status codes. | Key | Type | Default | Meaning | | -------------------- | ------ | -------- | ---------------------------------------------------------------------------------------- | | `path` | string | required | The path after `inbox.basePath`. | | `idempotencyKeyPath` | string | required | The JSONPath of the idempotency key. A request whose key path finds no value gets `400`. | | `auth` | block | none | The credentials that each request must carry. See the next section. | ### Source auth [Section titled “Source auth”](#source-auth) The same block configures `admin.auth`. A request that fails the check gets `401`. | Kind | Key | Type | Default | Meaning | | --------- | ------------------------ | ------------ | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `bearer` | `token` | secret | required | The request must send `Authorization: Bearer `. The scheme matches in any letter case. | | `api-key` | `key` | secret | required | The value of the key header. | | `api-key` | `headerName` | string | `X-API-Key` | The header that carries the key. | | `hmac` | `secret` | secret | required | The HMAC key. | | `hmac` | `headerName` | string | `X-Signature` | The header that carries the signature. | | `hmac` | `algorithm` | string | `HmacSHA256` | `HmacSHA256` or `HmacSHA512`. | | `hmac` | `signaturePrefix` | string | `sha256=` | The text before the hexadecimal signature in the header. | | `hmac` | `timestampHeader` | string | none | A header that carries the request time in Unix milliseconds. | | `hmac` | `timestampTolerance` | milliseconds | `300000` | The largest difference between the request time and the clock. It must be above 0. | | `hmac` | `signaturePayloadFormat` | enumeration | see below | `body` (or `BODY`) signs the body. `timestamp-dot-body` (or `TIMESTAMP_DOT_BODY`) signs the timestamp, a dot and the body. QueueBox 0.4.0 and earlier load only the upper-case names. | When `timestampHeader` is set, `signaturePayloadFormat` defaults to `TIMESTAMP_DOT_BODY`. Otherwise it defaults to `BODY`. `TIMESTAMP_DOT_BODY` without a `timestampHeader` stops the start. ### RabbitMQ source [Section titled “RabbitMQ source”](#rabbitmq-source) QueueBox acknowledges a delivery after the inbox row commits. A body that is not JSON, or a message that the transform rejects, becomes a dead row, and QueueBox acknowledges it. | Key | Type | Default | Meaning | | --------------------- | ------- | ----------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------- | | `queueName` | string | required | The queue to consume. | | `connectionUrl` | string | required | The AMQP URI of the broker. | | `prefetchCount` | integer | `10` | The largest number of unacknowledged deliveries. | | `declareQueue` | boolean | `false` | Declare the queue as durable before the consumer starts. Keep `false` when the queue exists, so a typo in `queueName` fails the start. | | `eventTypeFromHeader` | boolean | `false` | Declare that every publisher sets the event type header. | | `attributeHeaders` | block | see [attributeHeaders](#attributeheaders) | The header names of the three attributes. | The idempotency key comes from the first of these that gives a value: the key header, `idempotencyKeyPath`, the AMQP `messageId` property, a SHA-256 digest of the body. Two different messages with the same body and no other key deduplicate to one row. ### Kafka source [Section titled “Kafka source”](#kafka-source) The consumer commits an offset only after the inbox row commits. Every replica with the same `groupId` shares the partitions. | Key | Type | Default | Meaning | | --------------------- | --------------- | ----------------------------------------- | --------------------------------------------------------------------- | | `bootstrapServers` | string | required | The brokers, separated by commas. | | `topics` | list of strings | required | The Kafka topics. One consumer reads them all. No entry may be blank. | | `groupId` | string | required | The consumer group. | | `autoOffsetReset` | string | `earliest` | Where a new group starts: `earliest` or `latest`. | | `maxPollRecords` | integer | `100` | The largest number of records in one poll. It must be above 0. | | `eventTypeFromHeader` | boolean | `false` | Declare that every producer sets the event type header. | | `securityProtocol` | string | `PLAINTEXT` | `PLAINTEXT`, `SSL`, `SASL_PLAINTEXT` or `SASL_SSL`. | | `saslMechanism` | string | none | Required by a `SASL_` protocol. | | `saslUsername` | string | none | Required by a `SASL_` protocol. | | `saslPassword` | secret | none | Required by a `SASL_` protocol. | | `attributeHeaders` | block | see [attributeHeaders](#attributeheaders) | The header names of the three attributes. | The idempotency key comes from the first of these that gives a value: the key header, `idempotencyKeyPath`, the record key, a SHA-256 digest of the body. The aggregate identifier falls back to the record key when the body path and the header give nothing. ### NATS source [Section titled “NATS source”](#nats-source) The source reads a JetStream stream through a durable consumer. QueueBox never creates the stream. | Key | Type | Default | Meaning | | --------------------- | ------------ | ----------------------------------------- | ---------------------------------------------------------------------- | | `servers` | string | required | The server URLs, separated by commas. | | `stream` | string | required | The JetStream stream. | | `durable` | string | required | The durable consumer name. Replicas with the same name share the work. | | `filterSubject` | string | none | A subject filter. None reads every subject of the stream. | | `ackWaitMs` | milliseconds | `30000` | How long JetStream waits for the acknowledgement before it redelivers. | | `batchSize` | integer | `100` | The largest number of messages in one fetch. | | `eventTypeFromHeader` | boolean | `false` | Declare that every publisher sets the event type header. | | `username` | string | none | A user name. It needs `password`. | | `password` | secret | none | A password. It needs `username`. | | `token` | secret | none | A token. Set a token or a user name, not both. | | `attributeHeaders` | block | see [attributeHeaders](#attributeheaders) | The header names of the three attributes. | The idempotency key comes from the first of these that gives a value: the key header, `idempotencyKeyPath`, the `Nats-Msg-Id` header, a SHA-256 digest of the body. ### attributeHeaders [Section titled “attributeHeaders”](#attributeheaders) The RabbitMQ, Kafka and NATS sources read three attributes from message headers. These keys set the header names. | Key | Type | Default | Meaning | | --------------------------------- | ------ | ------------------- | ------------------------------------------------------------------------- | | `attributeHeaders.idempotencyKey` | string | `x-idempotency-key` | The header of the idempotency key. It comes before `idempotencyKeyPath`. | | `attributeHeaders.aggregateId` | string | `x-aggregate-id` | The header of the aggregate identifier. It comes after `aggregateIdPath`. | | `attributeHeaders.eventType` | string | `x-event-type` | The header of the event type. It comes after `eventTypePath`. | The [headers](/reference/headers/#headers-that-queuebox-reads) page gives the full order for each kind. ### filter [Section titled “filter”](#filter) A filter reads the headers of each received message. It applies to every kind of source. | Key | Type | Default | Meaning | | ---------------- | ------------- | ------- | -------------------------------- | | `filter.require` | list of rules | empty | A message must match every rule. | | `filter.exclude` | list of rules | empty | A message must match no rule. | | Rule key | Type | Meaning | | --------- | --------------- | ---------------------------------------------------------------------------------------- | | `header` | string | The header name. Required. It matches in any letter case. | | `equals` | string | The value must equal this string. | | `in` | list of strings | The value must equal one entry. The list must not be empty. | | `matches` | string | The value must match this topic glob. `*` matches one segment and `**` matches anything. | | `exists` | boolean | `true`: the header must be present. `false` stops the start. | A rule sets exactly one of `equals`, `in`, `matches` and `exists`. A value matches in the same letter case only. A rule with `equals`, `in` or `matches` does not match a message without the header. A message that does not pass leaves no row. A broker source acknowledges it. An HTTP source answers `202` with `{"status":"filtered"}`. The counter `queuebox_inbox_filtered_total` counts it. ## transform [Section titled “transform”](#transform) A `transform` block can sit on a source, a route and a destination. The [transforms](/reference/transforms/) page gives the variables of each stage. | Key | Type | Default | Meaning | | ------------ | ------------ | -------- | ------------------------------------------------------------------------------------------ | | `expression` | string | required | The JSONata expression. It must compile at start. | | `timeoutMs` | milliseconds | `100` | The longest run time of one evaluation. It must be above 0. | | `maxDepth` | integer | `100` | The deepest recursion of one evaluation. It must be above 0. | | `onError` | enumeration | `Fail` | `Fail`, `Skip` or `Dead`. See [error strategies](/reference/transforms/#error-strategies). | ## retention [Section titled “retention”](#retention) Retention deletes old rows in batches. It never deletes a row that waits or runs. | Key | Type | Default | Meaning | | -------------------- | ------- | -------- | --------------------------------------------------------------------- | | `retention.enabled` | boolean | `false` | Run the cleanup. The table blocks below apply only when it is `true`. | | `retention.outbox.*` | block | disabled | The outbox policy. | | `retention.inbox.*` | block | disabled | The inbox policy. | ### retention.outbox and retention.inbox [Section titled “retention.outbox and retention.inbox”](#retentionoutbox-and-retentioninbox) | Key | Type | Default | Meaning | | ----------------- | ------------------ | ---------- | --------------------------------------------------------------------------------- | | `policy` | enumeration | `DISABLED` | `AGE`, `COUNT` or `DISABLED`. The inbox accepts `AGE` or `DISABLED`. | | `maxAge` | retention duration | none | Required by `AGE`. Rows older than this go. | | `maxCount` | integer | none | Required by `COUNT`. It must be above 0. | | `cleanupInterval` | retention duration | `1h` | The wait between two cleanup runs. | | `batchSize` | integer | `1000` | The largest number of rows that one delete statement removes. It must be above 0. | The outbox age policy measures from `updated_at`. It deletes rows in state `sent` or `dead`. The inbox age policy measures from `created_at`. It deletes rows in state `processed` or `dead`. The outbox count policy keeps `maxCount` sent rows and, apart from them, `maxCount` dead rows. Caution A deleted inbox row no longer deduplicates. Set `retention.inbox.maxAge` above the longest time in which a sender can repeat a message. # Environment variables > How a QUEUEBOX_ environment variable maps to a configuration key, which source wins, and the variable of every key. This page describes how a `QUEUEBOX_` environment variable sets a configuration key, and lists the variable of every key. The [configuration](/reference/configuration/) page gives the type, the default and the meaning of each key. ## Precedence [Section titled “Precedence”](#precedence) QueueBox reads three sources. For each key, the first source in this list that sets the key wins. 1. The `QUEUEBOX_` environment variables. 2. The YAML file that `QUEUEBOX_CONFIG_FILE` names. When the variable is absent, QueueBox reads `/etc/queuebox/queuebox.yml`. 3. The `queuebox.yml` resource in the image. Source 2 and source 3 do not merge. When the file exists, QueueBox does not read the resource. When the file is absent and any `QUEUEBOX_` variable is set, QueueBox does not read the resource either. A deployment that uses variables only therefore gets no source, destination or route from the image. `QUEUEBOX_CONFIG_FILE` is not a configuration key. It only names the file. A variable merges with the file key by key inside a map. `QUEUEBOX_DESTINATIONS_ORDERS_TIMEOUTMS` changes the timeout of the `orders` destination, and the other keys of that destination stay as the file sets them. A list does not merge. A variable that sets one element of a list replaces the whole list of the file. Set every key of every element, or set the list in the file only. When no source holds a configuration, the start fails. The error names the three sources. ## Naming rule [Section titled “Naming rule”](#naming-rule) `EnvConfigLoader.envKeyToYamlPath` turns a variable name into a key path: 1. It removes the `QUEUEBOX_` prefix. 2. It turns a double underscore `__` into a literal `_`. 3. It turns every other `_` into a `.` between two levels. 4. It writes the path in lower case. The loader matches a key name in any letter case. The leaf name therefore has no underscore inside it. `outbox.pollIntervalMs` is `QUEUEBOX_OUTBOX_POLLINTERVALMS`. A variable that ends in `OUTBOX_POLL_INTERVAL_MS` sets the path `outbox.poll.interval.ms`, which does not exist. QueueBox reports no error for it, and the value has no effect. | Part of the path | How to write it | Example variable | Key | | ---------------------- | ----------------------------------- | --------------------------------------- | ----------------------------- | | A level | One `_` | `QUEUEBOX_DATABASE_URL` | `database.url` | | A camel case name | The name in upper case, with no `_` | `QUEUEBOX_SERVER_HTTPPORT` | `server.httpPort` | | A map key | The key in upper case | `QUEUEBOX_SOURCES_STRIPE_PATH` | `sources.stripe.path` | | A `_` inside a map key | `__` | `QUEUEBOX_DESTINATIONS_MY__API_BASEURL` | `destinations.my_api.baseUrl` | | A list index | The number | `QUEUEBOX_ROUTES_0_TOPICPATTERN` | `routes[0].topicPattern` | | A list of values | One index per value | `QUEUEBOX_SOURCES_ORDERS_TOPICS_0` | `sources.orders.topics[0]` | A map key that a variable sets is always lower case. A variable cannot set a map key that holds a `-` in most shells. Give a destination or a source a lower case name without `-` when a variable must reach it. Every validation error prints the variable name of the key, through `EnvConfigLoader.yamlPathToEnvKey`. ## Example [Section titled “Example”](#example) This set of variables configures one HTTP destination, one route and one HTTP source, with no file. ```bash QUEUEBOX_DATABASE_URL=jdbc:postgresql://db:5432/queuebox QUEUEBOX_DATABASE_USERNAME=queuebox QUEUEBOX_DATABASE_PASSWORD=secret QUEUEBOX_DESTINATIONS_WEBHOOK_TYPE=http QUEUEBOX_DESTINATIONS_WEBHOOK_BASEURL=https://api.example.com QUEUEBOX_ROUTES_0_TOPICPATTERN=order.* QUEUEBOX_ROUTES_0_DESTINATION=webhook QUEUEBOX_SOURCES_STRIPE_TYPE=http QUEUEBOX_SOURCES_STRIPE_PATH=/stripe QUEUEBOX_SOURCES_STRIPE_IDEMPOTENCYKEYPATH=$.id QUEUEBOX_SOURCES_STRIPE_EVENTTYPEPATH=$.type ``` A header filter uses list indexes at two levels: ```bash QUEUEBOX_SOURCES_ORDERS_FILTER_REQUIRE_0_HEADER=x-region QUEUEBOX_SOURCES_ORDERS_FILTER_REQUIRE_0_IN_0=eu QUEUEBOX_SOURCES_ORDERS_FILTER_REQUIRE_0_IN_1=uk QUEUEBOX_SOURCES_ORDERS_FILTER_EXCLUDE_0_HEADER=x-test QUEUEBOX_SOURCES_ORDERS_FILTER_EXCLUDE_0_EXISTS=true ``` The three database variables are the minimum. Every other key has a default or is optional. ## Every variable [Section titled “Every variable”](#every-variable) The tables use `orders` as the map key of a destination or a source, and `0` as a list index. Replace them with your own name and index. A key of a destination or a source applies only to the kinds that the [configuration](/reference/configuration/) page lists for it. ### server [Section titled “server”](#server) | Key | Variable | | ----------------------- | -------------------------------- | | `server.httpPort` | `QUEUEBOX_SERVER_HTTPPORT` | | `server.managementPort` | `QUEUEBOX_SERVER_MANAGEMENTPORT` | ### database [Section titled “database”](#database) | Key | Variable | | ------------------------------ | --------------------------------------- | | `database.type` | `QUEUEBOX_DATABASE_TYPE` | | `database.url` | `QUEUEBOX_DATABASE_URL` | | `database.username` | `QUEUEBOX_DATABASE_USERNAME` | | `database.password` | `QUEUEBOX_DATABASE_PASSWORD` | | `database.poolSize` | `QUEUEBOX_DATABASE_POOLSIZE` | | `database.connectionTimeoutMs` | `QUEUEBOX_DATABASE_CONNECTIONTIMEOUTMS` | | `database.outboxTableName` | `QUEUEBOX_DATABASE_OUTBOXTABLENAME` | | `database.inboxTableName` | `QUEUEBOX_DATABASE_INBOXTABLENAME` | | `database.migrate` | `QUEUEBOX_DATABASE_MIGRATE` | | `database.startupTimeoutMs` | `QUEUEBOX_DATABASE_STARTUPTIMEOUTMS` | ### database.columnMapping [Section titled “database.columnMapping”](#databasecolumnmapping) | Key | Variable | | ---------------------------------------------- | ------------------------------------------------------- | | `database.columnMapping.outbox.id` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_ID` | | `database.columnMapping.outbox.topic` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_TOPIC` | | `database.columnMapping.outbox.key` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_KEY` | | `database.columnMapping.outbox.aggregateType` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_AGGREGATETYPE` | | `database.columnMapping.outbox.payload` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_PAYLOAD` | | `database.columnMapping.outbox.headers` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_HEADERS` | | `database.columnMapping.outbox.state` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_STATE` | | `database.columnMapping.outbox.attempt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_ATTEMPT` | | `database.columnMapping.outbox.maxAttempts` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_MAXATTEMPTS` | | `database.columnMapping.outbox.scheduledAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_SCHEDULEDAT` | | `database.columnMapping.outbox.createdAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_CREATEDAT` | | `database.columnMapping.outbox.updatedAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_UPDATEDAT` | | `database.columnMapping.outbox.claimedAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_CLAIMEDAT` | | `database.columnMapping.outbox.claimToken` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_CLAIMTOKEN` | | `database.columnMapping.outbox.leaseExpiresAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_LEASEEXPIRESAT` | | `database.columnMapping.outbox.lastError` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_LASTERROR` | | `database.columnMapping.outbox.sequence` | `QUEUEBOX_DATABASE_COLUMNMAPPING_OUTBOX_SEQUENCE` | | `database.columnMapping.inbox.id` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_ID` | | `database.columnMapping.inbox.source` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_SOURCE` | | `database.columnMapping.inbox.idempotencyKey` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_IDEMPOTENCYKEY` | | `database.columnMapping.inbox.aggregateId` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_AGGREGATEID` | | `database.columnMapping.inbox.eventType` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_EVENTTYPE` | | `database.columnMapping.inbox.payload` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_PAYLOAD` | | `database.columnMapping.inbox.state` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_STATE` | | `database.columnMapping.inbox.createdAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_CREATEDAT` | | `database.columnMapping.inbox.processedAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_PROCESSEDAT` | | `database.columnMapping.inbox.claimedAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_CLAIMEDAT` | | `database.columnMapping.inbox.claimToken` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_CLAIMTOKEN` | | `database.columnMapping.inbox.leaseExpiresAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_LEASEEXPIRESAT` | | `database.columnMapping.inbox.correlationId` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_CORRELATIONID` | | `database.columnMapping.inbox.consumption` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_CONSUMPTION` | | `database.columnMapping.inbox.scheduledAt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_SCHEDULEDAT` | | `database.columnMapping.inbox.attempt` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_ATTEMPT` | | `database.columnMapping.inbox.lastError` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_LASTERROR` | | `database.columnMapping.inbox.headers` | `QUEUEBOX_DATABASE_COLUMNMAPPING_INBOX_HEADERS` | ### outbox [Section titled “outbox”](#outbox) | Key | Variable | | -------------------------------------------------- | ----------------------------------------------------------- | | `outbox.capture.mode` | `QUEUEBOX_OUTBOX_CAPTURE_MODE` | | `outbox.capture.enabled` | `QUEUEBOX_OUTBOX_CAPTURE_ENABLED` | | `outbox.capture.identity` | `QUEUEBOX_OUTBOX_CAPTURE_IDENTITY` | | `outbox.capture.stateDirectory` | `QUEUEBOX_OUTBOX_CAPTURE_STATEDIRECTORY` | | `outbox.capture.schema` | `QUEUEBOX_OUTBOX_CAPTURE_SCHEMA` | | `outbox.capture.publication` | `QUEUEBOX_OUTBOX_CAPTURE_PUBLICATION` | | `outbox.capture.slot` | `QUEUEBOX_OUTBOX_CAPTURE_SLOT` | | `outbox.capture.reconciliationIntervalMs` | `QUEUEBOX_OUTBOX_CAPTURE_RECONCILIATIONINTERVALMS` | | `outbox.capture.connection.hostname` | `QUEUEBOX_OUTBOX_CAPTURE_CONNECTION_HOSTNAME` | | `outbox.capture.connection.port` | `QUEUEBOX_OUTBOX_CAPTURE_CONNECTION_PORT` | | `outbox.capture.connection.database` | `QUEUEBOX_OUTBOX_CAPTURE_CONNECTION_DATABASE` | | `outbox.capture.connection.username` | `QUEUEBOX_OUTBOX_CAPTURE_CONNECTION_USERNAME` | | `outbox.capture.connection.password` | `QUEUEBOX_OUTBOX_CAPTURE_CONNECTION_PASSWORD` | | `outbox.capture.connection.encrypt` | `QUEUEBOX_OUTBOX_CAPTURE_CONNECTION_ENCRYPT` | | `outbox.capture.connection.trustServerCertificate` | `QUEUEBOX_OUTBOX_CAPTURE_CONNECTION_TRUSTSERVERCERTIFICATE` | | `outbox.pollIntervalMs` | `QUEUEBOX_OUTBOX_POLLINTERVALMS` | | `outbox.batchSize` | `QUEUEBOX_OUTBOX_BATCHSIZE` | | `outbox.retryBaseDelayMs` | `QUEUEBOX_OUTBOX_RETRYBASEDELAYMS` | | `outbox.maxAttempts` | `QUEUEBOX_OUTBOX_MAXATTEMPTS` | | `outbox.claimTimeoutMs` | `QUEUEBOX_OUTBOX_CLAIMTIMEOUTMS` | | `outbox.concurrency` | `QUEUEBOX_OUTBOX_CONCURRENCY` | | `outbox.pendingGaugeIntervalMs` | `QUEUEBOX_OUTBOX_PENDINGGAUGEINTERVALMS` | | `outbox.shutdownTimeoutMs` | `QUEUEBOX_OUTBOX_SHUTDOWNTIMEOUTMS` | ### inbox [Section titled “inbox”](#inbox) | Key | Variable | | ------------------------------------ | --------------------------------------------- | | `inbox.basePath` | `QUEUEBOX_INBOX_BASEPATH` | | `inbox.relay.enabled` | `QUEUEBOX_INBOX_RELAY_ENABLED` | | `inbox.relay.pollIntervalMs` | `QUEUEBOX_INBOX_RELAY_POLLINTERVALMS` | | `inbox.relay.batchSize` | `QUEUEBOX_INBOX_RELAY_BATCHSIZE` | | `inbox.relay.claimTimeoutMs` | `QUEUEBOX_INBOX_RELAY_CLAIMTIMEOUTMS` | | `inbox.relay.pendingGaugeIntervalMs` | `QUEUEBOX_INBOX_RELAY_PENDINGGAUGEINTERVALMS` | | `inbox.relay.maxAttempts` | `QUEUEBOX_INBOX_RELAY_MAXATTEMPTS` | | `inbox.maxBodyBytes` | `QUEUEBOX_INBOX_MAXBODYBYTES` | ### http [Section titled “http”](#http) | Key | Variable | | ---------------------------- | ------------------------------------- | | `http.maxErrorBodyBytes` | `QUEUEBOX_HTTP_MAXERRORBODYBYTES` | | `http.blockPrivateAddresses` | `QUEUEBOX_HTTP_BLOCKPRIVATEADDRESSES` | ### admin [Section titled “admin”](#admin) | Key | Variable | | ----------------------------------- | -------------------------------------------- | | `admin.enabled` | `QUEUEBOX_ADMIN_ENABLED` | | `admin.insecure` | `QUEUEBOX_ADMIN_INSECURE` | | `admin.auth.type` | `QUEUEBOX_ADMIN_AUTH_TYPE` | | `admin.auth.headerName` | `QUEUEBOX_ADMIN_AUTH_HEADERNAME` | | `admin.auth.key` | `QUEUEBOX_ADMIN_AUTH_KEY` | | `admin.auth.token` | `QUEUEBOX_ADMIN_AUTH_TOKEN` | | `admin.auth.secret` | `QUEUEBOX_ADMIN_AUTH_SECRET` | | `admin.auth.algorithm` | `QUEUEBOX_ADMIN_AUTH_ALGORITHM` | | `admin.auth.signaturePrefix` | `QUEUEBOX_ADMIN_AUTH_SIGNATUREPREFIX` | | `admin.auth.timestampHeader` | `QUEUEBOX_ADMIN_AUTH_TIMESTAMPHEADER` | | `admin.auth.timestampTolerance` | `QUEUEBOX_ADMIN_AUTH_TIMESTAMPTOLERANCE` | | `admin.auth.signaturePayloadFormat` | `QUEUEBOX_ADMIN_AUTH_SIGNATUREPAYLOADFORMAT` | | `admin.maxTransformTimeoutMs` | `QUEUEBOX_ADMIN_MAXTRANSFORMTIMEOUTMS` | | `admin.maxPayloadBytes` | `QUEUEBOX_ADMIN_MAXPAYLOADBYTES` | ### destinations [Section titled “destinations”](#destinations) | Key | Variable | | --------------------------------------------- | -------------------------------------------------------- | | `destinations..type` | `QUEUEBOX_DESTINATIONS_ORDERS_TYPE` | | `destinations..baseUrl` | `QUEUEBOX_DESTINATIONS_ORDERS_BASEURL` | | `destinations..path` | `QUEUEBOX_DESTINATIONS_ORDERS_PATH` | | `destinations..timeoutMs` | `QUEUEBOX_DESTINATIONS_ORDERS_TIMEOUTMS` | | `destinations..headers.` | `QUEUEBOX_DESTINATIONS_ORDERS_HEADERS_XTEAM` | | `destinations..transform.expression` | `QUEUEBOX_DESTINATIONS_ORDERS_TRANSFORM_EXPRESSION` | | `destinations..transform.timeoutMs` | `QUEUEBOX_DESTINATIONS_ORDERS_TRANSFORM_TIMEOUTMS` | | `destinations..transform.maxDepth` | `QUEUEBOX_DESTINATIONS_ORDERS_TRANSFORM_MAXDEPTH` | | `destinations..transform.onError` | `QUEUEBOX_DESTINATIONS_ORDERS_TRANSFORM_ONERROR` | | `destinations..auth.type` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_TYPE` | | `destinations..auth.username` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_USERNAME` | | `destinations..auth.password` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_PASSWORD` | | `destinations..auth.headerName` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_HEADERNAME` | | `destinations..auth.headerValue` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_HEADERVALUE` | | `destinations..auth.clientId` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_CLIENTID` | | `destinations..auth.clientSecret` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_CLIENTSECRET` | | `destinations..auth.tokenUrl` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_TOKENURL` | | `destinations..auth.scope` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_SCOPE` | | `destinations..auth.extraParams.` | `QUEUEBOX_DESTINATIONS_ORDERS_AUTH_EXTRAPARAMS_AUDIENCE` | | `destinations..bootstrapServers` | `QUEUEBOX_DESTINATIONS_ORDERS_BOOTSTRAPSERVERS` | | `destinations..topic` | `QUEUEBOX_DESTINATIONS_ORDERS_TOPIC` | | `destinations..keyTemplate` | `QUEUEBOX_DESTINATIONS_ORDERS_KEYTEMPLATE` | | `destinations..securityProtocol` | `QUEUEBOX_DESTINATIONS_ORDERS_SECURITYPROTOCOL` | | `destinations..saslMechanism` | `QUEUEBOX_DESTINATIONS_ORDERS_SASLMECHANISM` | | `destinations..saslUsername` | `QUEUEBOX_DESTINATIONS_ORDERS_SASLUSERNAME` | | `destinations..saslPassword` | `QUEUEBOX_DESTINATIONS_ORDERS_SASLPASSWORD` | | `destinations..topicFrom` | `QUEUEBOX_DESTINATIONS_ORDERS_TOPICFROM` | | `destinations..servers` | `QUEUEBOX_DESTINATIONS_ORDERS_SERVERS` | | `destinations..subject` | `QUEUEBOX_DESTINATIONS_ORDERS_SUBJECT` | | `destinations..jetStream` | `QUEUEBOX_DESTINATIONS_ORDERS_JETSTREAM` | | `destinations..username` | `QUEUEBOX_DESTINATIONS_ORDERS_USERNAME` | | `destinations..password` | `QUEUEBOX_DESTINATIONS_ORDERS_PASSWORD` | | `destinations..token` | `QUEUEBOX_DESTINATIONS_ORDERS_TOKEN` | | `destinations..subjectFrom` | `QUEUEBOX_DESTINATIONS_ORDERS_SUBJECTFROM` | | `destinations..url` | `QUEUEBOX_DESTINATIONS_ORDERS_URL` | | `destinations..exchange` | `QUEUEBOX_DESTINATIONS_ORDERS_EXCHANGE` | | `destinations..exchangeType` | `QUEUEBOX_DESTINATIONS_ORDERS_EXCHANGETYPE` | | `destinations..routingKeyTemplate` | `QUEUEBOX_DESTINATIONS_ORDERS_ROUTINGKEYTEMPLATE` | | `destinations..exchangeFrom` | `QUEUEBOX_DESTINATIONS_ORDERS_EXCHANGEFROM` | | `destinations..deliveryMode` | `QUEUEBOX_DESTINATIONS_ORDERS_DELIVERYMODE` | ### routes [Section titled “routes”](#routes) | Key | Variable | | ------------------------------------------ | ------------------------------------------------- | | `routes..topicPattern` | `QUEUEBOX_ROUTES_0_TOPICPATTERN` | | `routes..destination` | `QUEUEBOX_ROUTES_0_DESTINATION` | | `routes..routingKeyTemplate` | `QUEUEBOX_ROUTES_0_ROUTINGKEYTEMPLATE` | | `routes..routingKeyMissingFieldDefault` | `QUEUEBOX_ROUTES_0_ROUTINGKEYMISSINGFIELDDEFAULT` | | `routes..transform.expression` | `QUEUEBOX_ROUTES_0_TRANSFORM_EXPRESSION` | | `routes..transform.timeoutMs` | `QUEUEBOX_ROUTES_0_TRANSFORM_TIMEOUTMS` | | `routes..transform.maxDepth` | `QUEUEBOX_ROUTES_0_TRANSFORM_MAXDEPTH` | | `routes..transform.onError` | `QUEUEBOX_ROUTES_0_TRANSFORM_ONERROR` | ### sources [Section titled “sources”](#sources) | Key | Variable | | ------------------------------------------------ | --------------------------------------------------------- | | `sources..type` | `QUEUEBOX_SOURCES_ORDERS_TYPE` | | `sources..path` | `QUEUEBOX_SOURCES_ORDERS_PATH` | | `sources..idempotencyKeyPath` | `QUEUEBOX_SOURCES_ORDERS_IDEMPOTENCYKEYPATH` | | `sources..aggregateIdPath` | `QUEUEBOX_SOURCES_ORDERS_AGGREGATEIDPATH` | | `sources..eventTypePath` | `QUEUEBOX_SOURCES_ORDERS_EVENTTYPEPATH` | | `sources..transform.expression` | `QUEUEBOX_SOURCES_ORDERS_TRANSFORM_EXPRESSION` | | `sources..transform.timeoutMs` | `QUEUEBOX_SOURCES_ORDERS_TRANSFORM_TIMEOUTMS` | | `sources..transform.maxDepth` | `QUEUEBOX_SOURCES_ORDERS_TRANSFORM_MAXDEPTH` | | `sources..transform.onError` | `QUEUEBOX_SOURCES_ORDERS_TRANSFORM_ONERROR` | | `sources..topic` | `QUEUEBOX_SOURCES_ORDERS_TOPIC` | | `sources..consumption` | `QUEUEBOX_SOURCES_ORDERS_CONSUMPTION` | | `sources..rateLimit.requestsPerMinute` | `QUEUEBOX_SOURCES_ORDERS_RATELIMIT_REQUESTSPERMINUTE` | | `sources..filter.require..header` | `QUEUEBOX_SOURCES_ORDERS_FILTER_REQUIRE_0_HEADER` | | `sources..filter.require..equals` | `QUEUEBOX_SOURCES_ORDERS_FILTER_REQUIRE_0_EQUALS` | | `sources..filter.require..in.` | `QUEUEBOX_SOURCES_ORDERS_FILTER_REQUIRE_0_IN_0` | | `sources..filter.require..matches` | `QUEUEBOX_SOURCES_ORDERS_FILTER_REQUIRE_0_MATCHES` | | `sources..filter.require..exists` | `QUEUEBOX_SOURCES_ORDERS_FILTER_REQUIRE_0_EXISTS` | | `sources..filter.exclude..header` | `QUEUEBOX_SOURCES_ORDERS_FILTER_EXCLUDE_0_HEADER` | | `sources..filter.exclude..equals` | `QUEUEBOX_SOURCES_ORDERS_FILTER_EXCLUDE_0_EQUALS` | | `sources..filter.exclude..in.` | `QUEUEBOX_SOURCES_ORDERS_FILTER_EXCLUDE_0_IN_0` | | `sources..filter.exclude..matches` | `QUEUEBOX_SOURCES_ORDERS_FILTER_EXCLUDE_0_MATCHES` | | `sources..filter.exclude..exists` | `QUEUEBOX_SOURCES_ORDERS_FILTER_EXCLUDE_0_EXISTS` | | `sources..auth.type` | `QUEUEBOX_SOURCES_ORDERS_AUTH_TYPE` | | `sources..auth.headerName` | `QUEUEBOX_SOURCES_ORDERS_AUTH_HEADERNAME` | | `sources..auth.key` | `QUEUEBOX_SOURCES_ORDERS_AUTH_KEY` | | `sources..auth.token` | `QUEUEBOX_SOURCES_ORDERS_AUTH_TOKEN` | | `sources..auth.secret` | `QUEUEBOX_SOURCES_ORDERS_AUTH_SECRET` | | `sources..auth.algorithm` | `QUEUEBOX_SOURCES_ORDERS_AUTH_ALGORITHM` | | `sources..auth.signaturePrefix` | `QUEUEBOX_SOURCES_ORDERS_AUTH_SIGNATUREPREFIX` | | `sources..auth.timestampHeader` | `QUEUEBOX_SOURCES_ORDERS_AUTH_TIMESTAMPHEADER` | | `sources..auth.timestampTolerance` | `QUEUEBOX_SOURCES_ORDERS_AUTH_TIMESTAMPTOLERANCE` | | `sources..auth.signaturePayloadFormat` | `QUEUEBOX_SOURCES_ORDERS_AUTH_SIGNATUREPAYLOADFORMAT` | | `sources..bootstrapServers` | `QUEUEBOX_SOURCES_ORDERS_BOOTSTRAPSERVERS` | | `sources..topics.` | `QUEUEBOX_SOURCES_ORDERS_TOPICS_0` | | `sources..groupId` | `QUEUEBOX_SOURCES_ORDERS_GROUPID` | | `sources..eventTypeFromHeader` | `QUEUEBOX_SOURCES_ORDERS_EVENTTYPEFROMHEADER` | | `sources..autoOffsetReset` | `QUEUEBOX_SOURCES_ORDERS_AUTOOFFSETRESET` | | `sources..maxPollRecords` | `QUEUEBOX_SOURCES_ORDERS_MAXPOLLRECORDS` | | `sources..securityProtocol` | `QUEUEBOX_SOURCES_ORDERS_SECURITYPROTOCOL` | | `sources..saslMechanism` | `QUEUEBOX_SOURCES_ORDERS_SASLMECHANISM` | | `sources..saslUsername` | `QUEUEBOX_SOURCES_ORDERS_SASLUSERNAME` | | `sources..saslPassword` | `QUEUEBOX_SOURCES_ORDERS_SASLPASSWORD` | | `sources..attributeHeaders.idempotencyKey` | `QUEUEBOX_SOURCES_ORDERS_ATTRIBUTEHEADERS_IDEMPOTENCYKEY` | | `sources..attributeHeaders.aggregateId` | `QUEUEBOX_SOURCES_ORDERS_ATTRIBUTEHEADERS_AGGREGATEID` | | `sources..attributeHeaders.eventType` | `QUEUEBOX_SOURCES_ORDERS_ATTRIBUTEHEADERS_EVENTTYPE` | | `sources..servers` | `QUEUEBOX_SOURCES_ORDERS_SERVERS` | | `sources..stream` | `QUEUEBOX_SOURCES_ORDERS_STREAM` | | `sources..durable` | `QUEUEBOX_SOURCES_ORDERS_DURABLE` | | `sources..filterSubject` | `QUEUEBOX_SOURCES_ORDERS_FILTERSUBJECT` | | `sources..ackWaitMs` | `QUEUEBOX_SOURCES_ORDERS_ACKWAITMS` | | `sources..batchSize` | `QUEUEBOX_SOURCES_ORDERS_BATCHSIZE` | | `sources..username` | `QUEUEBOX_SOURCES_ORDERS_USERNAME` | | `sources..password` | `QUEUEBOX_SOURCES_ORDERS_PASSWORD` | | `sources..token` | `QUEUEBOX_SOURCES_ORDERS_TOKEN` | | `sources..queueName` | `QUEUEBOX_SOURCES_ORDERS_QUEUENAME` | | `sources..connectionUrl` | `QUEUEBOX_SOURCES_ORDERS_CONNECTIONURL` | | `sources..prefetchCount` | `QUEUEBOX_SOURCES_ORDERS_PREFETCHCOUNT` | | `sources..declareQueue` | `QUEUEBOX_SOURCES_ORDERS_DECLAREQUEUE` | ### retention [Section titled “retention”](#retention) | Key | Variable | | ---------------------------------- | ------------------------------------------- | | `retention.enabled` | `QUEUEBOX_RETENTION_ENABLED` | | `retention.outbox.policy` | `QUEUEBOX_RETENTION_OUTBOX_POLICY` | | `retention.outbox.maxAge` | `QUEUEBOX_RETENTION_OUTBOX_MAXAGE` | | `retention.outbox.maxCount` | `QUEUEBOX_RETENTION_OUTBOX_MAXCOUNT` | | `retention.outbox.cleanupInterval` | `QUEUEBOX_RETENTION_OUTBOX_CLEANUPINTERVAL` | | `retention.outbox.batchSize` | `QUEUEBOX_RETENTION_OUTBOX_BATCHSIZE` | | `retention.inbox.policy` | `QUEUEBOX_RETENTION_INBOX_POLICY` | | `retention.inbox.maxAge` | `QUEUEBOX_RETENTION_INBOX_MAXAGE` | | `retention.inbox.maxCount` | `QUEUEBOX_RETENTION_INBOX_MAXCOUNT` | | `retention.inbox.cleanupInterval` | `QUEUEBOX_RETENTION_INBOX_CLEANUPINTERVAL` | | `retention.inbox.batchSize` | `QUEUEBOX_RETENTION_INBOX_BATCHSIZE` | # Headers > Every header that QueueBox sends on a delivery, and every header that it reads when a message arrives. This page lists every header that QueueBox sends when it delivers an outbox row, and every header that it reads when a source receives a message. ## Headers that QueueBox sends [Section titled “Headers that QueueBox sends”](#headers-that-queuebox-sends) A delivery carries three groups of headers: 1. The standard headers in the tables below. 2. The static `headers` of the destination, and on an HTTP destination the header of the `auth` block. 3. The entries of the `headers` column of the outbox row. A row header wins over a static header or an auth header of the same name. Do not give a row header the name of a standard header. QueueBox does not remove the standard header, so the message then carries two values. ### HTTP destination [Section titled “HTTP destination”](#http-destination) QueueBox sends a `POST` with the payload as the body. | Header | Value | | --------------- | ---------------------------------------------------------------------------------------------------------------------------------- | | `Content-Type` | `application/json` | | `X-Message-Id` | The `id` of the outbox row. It stays the same on every retry and on a replay of the row. | | `X-Topic` | The `topic` of the row. | | `X-Attempt` | The `attempt` of the row. It is `0` on the first delivery. | | `X-Message-Key` | The `key` of the row. It is absent when the row has no key. | | `Authorization` | `Bearer ` for `oauth2`, or `Basic ` for `basic`. The `header` kind of auth sends its own `headerName` instead. | ### RabbitMQ destination [Section titled “RabbitMQ destination”](#rabbitmq-destination) | Header or property | Value | | ------------------------ | -------------------------------------------------------- | | `message_id` property | The `id` of the outbox row. | | `content_type` property | `application/json` | | `delivery_mode` property | `2` for `deliveryMode: persistent`, `1` for `transient`. | | `x-topic` header | The `topic` of the row. | | `x-attempt` header | The `attempt` of the row, as an integer. | The routing key carries the rendered `routingKeyTemplate`. A RabbitMQ message has no key header. ### Kafka destination [Section titled “Kafka destination”](#kafka-destination) | Header or field | Value | | --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | | Record key | The route `routingKeyTemplate`, or else the destination `keyTemplate`. The default is the `key` of the row. An empty result sends no record key. | | `x-message-id` header | The `id` of the outbox row. | | `x-topic` header | The `topic` of the row. | | `x-attempt` header | The `attempt` of the row. | ### NATS destination [Section titled “NATS destination”](#nats-destination) | Header | Value | | --------------- | ----------------------------------------------------------- | | `x-message-id` | The `id` of the outbox row. | | `x-topic` | The `topic` of the row. | | `x-attempt` | The `attempt` of the row. | | `x-message-key` | The `key` of the row. It is absent when the row has no key. | ## Headers that the relay adds [Section titled “Headers that the relay adds”](#headers-that-the-relay-adds) The relay copies the `headers` of the inbox row onto the outbox row. Then it sets four headers, so a destination can trace the message to its source. | Header | Value | | ------------------- | ------------------------------------------------------------- | | `x-inbox-id` | The `id` of the inbox row. | | `x-source` | The source name. | | `x-idempotency-key` | The idempotency key of the inbox row. | | `X-Correlation-Id` | The correlation identifier of the inbox row, when it has one. | Each of these four replaces a received header of the same name in any letter case. A sender therefore cannot set them. ## Which identifier to deduplicate on [Section titled “Which identifier to deduplicate on”](#which-identifier-to-deduplicate-on) | Traffic | Deduplicate on | Why | | ---------------------------------- | ------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------- | | A row that the application inserts | `X-Message-Id` | One outbox row keeps one `id` through every retry. | | A row that the relay creates | `x-idempotency-key`, with `x-source` | The relay creates a new outbox row, with a new `id`, when an operator replays an inbox row. The idempotency key stays the same. | Delivery is at least once, so a receiver must deduplicate. [Delivery semantics](/concepts/delivery-semantics/) explains why a repeat can occur. ## Headers that QueueBox reads [Section titled “Headers that QueueBox reads”](#headers-that-queuebox-reads) ### HTTP source [Section titled “HTTP source”](#http-source) | Header | Use | | -------------------------- | -------------------------------------------------------------------------------------------------------------- | | `Authorization` | The `bearer` kind of source auth reads `Bearer `. | | `X-API-Key` | The `api-key` kind reads the key. `auth.headerName` changes the name. | | `X-Signature` | The `hmac` kind reads the signature. `auth.headerName` changes the name. | | The `auth.timestampHeader` | The `hmac` kind reads the request time in Unix milliseconds. | | `X-Correlation-Id` | The correlation identifier. QueueBox generates a UUID when it is absent, and echoes the value in the response. | | Any header | The header filter of the source reads it. QueueBox stores it in the inbox `headers` column. | QueueBox does not store `Authorization`, `Proxy-Authorization`, `Cookie`, or the header that the `api-key` or `hmac` auth reads. An HTTP source takes the idempotency key, the event type and the aggregate identifier from the body only. ### RabbitMQ, Kafka and NATS sources [Section titled “RabbitMQ, Kafka and NATS sources”](#rabbitmq-kafka-and-nats-sources) Each attribute comes from the first place in its list that gives a value. | Attribute | RabbitMQ | Kafka | NATS | | ---------------------- | ----------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------- | | Idempotency key | 1. `x-idempotency-key` header 2. `idempotencyKeyPath` 3. AMQP `message_id` property 4. SHA-256 digest of the body | 1. `x-idempotency-key` header 2. `idempotencyKeyPath` 3. record key 4. SHA-256 digest of the body | 1. `x-idempotency-key` header 2. `idempotencyKeyPath` 3. `Nats-Msg-Id` header 4. SHA-256 digest of the body | | Aggregate identifier | 1. `aggregateIdPath` 2. `x-aggregate-id` header | 1. `aggregateIdPath` 2. `x-aggregate-id` header 3. record key | 1. `aggregateIdPath` 2. `x-aggregate-id` header | | Event type | 1. `eventTypePath` 2. `x-event-type` header | 1. `eventTypePath` 2. `x-event-type` header | 1. `eventTypePath` 2. `x-event-type` header | | Correlation identifier | 1. `X-Correlation-Id` header, in any letter case 2. AMQP `correlation_id` property 3. a new UUID | 1. `X-Correlation-Id` header 2. a new UUID | 1. `X-Correlation-Id` header 2. a new UUID | `attributeHeaders.idempotencyKey`, `attributeHeaders.aggregateId` and `attributeHeaders.eventType` change the three `x-` header names. A Debezium producer, for example, sends `id`, `aggregateId` and `eventType`. The [configuration](/reference/configuration/#attributeheaders) page describes the keys. The digest fallback deduplicates a redelivery of the same bytes. It also merges two different events that carry the same body. Give each message a key through the header, the body path or the broker identifier. QueueBox removes control characters from a correlation identifier and keeps at most 128 characters. Every received header also goes to the inbox `headers` column and to the header filter. The [inbox table](/reference/inbox-table/#headers) page gives the conversion rules of each broker. ## Headers in a transform [Section titled “Headers in a transform”](#headers-in-a-transform) A source transform reads the received headers as `$headers`, one string value per name. A route or destination transform has no `$headers`. The [transforms](/reference/transforms/) page lists the variables of each stage. # HTTP API > Every HTTP route of QueueBox, with its method, authentication, request, status codes and response bodies. This page lists every HTTP route that QueueBox serves: the inbox sources, the health routes, the metrics route and the admin routes. ## Ports [Section titled “Ports”](#ports) | Route | Port without `server.managementPort` | Port with `server.managementPort` | | ------------------------------------------------------ | ------------------------------------ | --------------------------------- | | `GET /` | `server.httpPort` | `server.httpPort` | | `POST ` | `server.httpPort` | `server.httpPort` | | `GET /health/live`, `GET /health/ready`, `GET /health` | `server.httpPort` | `server.managementPort` only | | `GET /metrics` | `server.httpPort` | `server.managementPort` only | | `POST /admin/transform/test`, `POST /admin/replay` | `server.httpPort` | `server.managementPort` only | With a management port, the data port answers `404` on the health, metrics and admin paths. A path that no route serves answers `404`. Two rules apply to every route of the data port: * A request whose `Content-Length` is above `inbox.maxBodyBytes` gets `413` with `{"error":"Request body exceeds bytes"}`. * A request that arrives during shutdown gets `503` with `{"error":"QueueBox is shutting down"}`. ## Root [Section titled “Root”](#root) `GET /` answers `200` with the text `QueueBox is running!`. It checks nothing. Use the health routes for a probe. ## Inbox sources [Section titled “Inbox sources”](#inbox-sources) Each HTTP source of `sources` registers one route: `POST` on `inbox.basePath` followed by the `path` of the source. With the default `basePath`, a source with `path: /stripe` answers at `/inbox/stripe`. The source name is not part of the path. ```bash curl -X POST http://localhost:8080/inbox/stripe \ -H "Content-Type: application/json" \ -d '{"id": "evt_123", "type": "payment.completed"}' ``` ### Request [Section titled “Request”](#request) | Part | Rule | | ------------------ | ------------------------------------------------------------------------------------------------------------------------ | | Body | One JSON document, at most `inbox.maxBodyBytes` bytes. The route counts the bytes of a chunked body as it reads them. | | `Content-Type` | The route parses the body as JSON whatever the header says. | | Authentication | The `auth` block of the source, when it has one. See [source auth](/reference/configuration/#source-auth). | | `X-Correlation-Id` | Optional. QueueBox stores it on the row. Without it, QueueBox generates a UUID. | | Other headers | QueueBox stores them in the `headers` column, except the credential headers. The header filter of the source reads them. | ### Processing order [Section titled “Processing order”](#processing-order) 1. QueueBox reads the body under the size cap. 2. It checks the credentials. 3. It runs the header filter. 4. It sets the `X-Correlation-Id` response header. 5. It parses the body as JSON. 6. It extracts the idempotency key, the event type and the aggregate identifier. 7. It runs the source transform. 8. It stores the row. The unique constraint on `(source, idempotency_key)` detects a duplicate here. The transform runs before the duplicate check. A repeat of a stored message whose transform now fails therefore gets `422`, not `200`. ### Responses [Section titled “Responses”](#responses) | Status | Body | Meaning | | ------ | -------------------------------------------- | ---------------------------------------------------------------------------------------------------- | | `202` | `{"messageId":""}` | The message is new. QueueBox stored it. The value is the `id` of the inbox row. | | `202` | `{"status":"filtered"}` | The header filter dropped the message. QueueBox stored no row. | | `200` | `{"status":"duplicate"}` | The source already holds a message with this idempotency key. | | `400` | `{"error":"Invalid JSON"}` | The body is not JSON. | | `400` | `{"error":""}` | The idempotency key path found no value. | | `401` | `{"error":""}` | The request failed the authentication of the source. The reason names the check, never a credential. | | `413` | `{"error":"Request body exceeds bytes"}` | The body is larger than `inbox.maxBodyBytes`. | | `422` | `{"error":"Transform failed: "}` | The source transform rejected the payload. QueueBox stored no row. | | `429` | none | The request went over `rateLimit.requestsPerMinute`. The response carries `Retry-After`. | | `500` | `{"error":"Storage failed"}` | The database write failed. Send the message again. | | `503` | `{"error":"QueueBox is shutting down"}` | The process is stopping. Send the message again. | A `202` means that the message is durable in the inbox, not that a destination received it. A sender can send the message again after a `500` or a `503`. A `4xx` answer does not change when the sender repeats the same request. ## Health [Section titled “Health”](#health) The health routes need no authentication. Each answers JSON. ### GET /health/live [Section titled “GET /health/live”](#get-healthlive) Liveness. It reports the process and touches no dependency, so a slow database cannot fail it. Use it for a liveness probe. ```json {"status":"healthy","components":{"process":{"status":"up"}}} ``` It always answers `200`. ### GET /health/ready [Section titled “GET /health/ready”](#get-healthready) Readiness. It checks the database and every worker. Use it for a readiness probe. | Status | Body `status` | Meaning | | ------ | ------------- | -------------------------------------------------------- | | `200` | `healthy` | Every component that decides readiness is `up`. | | `503` | `unhealthy` | At least one component that decides readiness is `down`. | ```json { "status": "healthy", "components": { "database": {"status": "up"}, "outbox-poller": {"status": "up"}, "inbox-relay": {"status": "up"}, "rabbitmq.orders-queue": {"status": "up"} } } ``` | Component | Present when | Decides readiness | | ------------------- | ---------------------------------- | ---------------------------------------------------------------------------- | | `database` | always | yes | | `outbox-capture` | `outbox.capture.enabled` is `true` | no. The component shows a capture fault, and delivery continues through SQL. | | `outbox-poller` | always | yes | | `retention-service` | `retention.enabled` is `true` | yes | | `inbox-relay` | `inbox.relay.enabled` is `true` | yes | | `rabbitmq.` | one per RabbitMQ source | yes | | `kafka.` | one per Kafka source | yes | | `nats.` | one per NATS source | yes | Each check has a bound of 3 seconds. A check that does not answer in time counts as `down`. ### GET /health [Section titled “GET /health”](#get-health) An alias of `GET /health/ready`, with the same status codes and body. ## Metrics [Section titled “Metrics”](#metrics) `GET /metrics` answers `200` with the Prometheus text format, content type `text/plain; version=0.0.4; charset=utf-8`. It needs no authentication. Keep it on an internal network, because the names and counts show destinations and traffic. The [metrics](/reference/metrics/) page lists every metric. ## Admin routes [Section titled “Admin routes”](#admin-routes) The admin routes exist only when `admin.enabled` is `true`. Otherwise they answer `404`. Each request needs the credentials of `admin.auth`, unless `admin.insecure` is `true`. The body is JSON of at most `admin.maxPayloadBytes` bytes. The route ignores an unknown field. ### POST /admin/transform/test [Section titled “POST /admin/transform/test”](#post-admintransformtest) Evaluates a JSONata expression against a sample payload. It sends nothing and writes nothing. | Field | Type | Required | Meaning | | ------------ | ------- | -------- | -------------------------------------------------------------------------------------------- | | `expression` | string | yes | The JSONata expression. | | `payload` | JSON | yes | The input of the expression. | | `mockTopic` | string | no | The value of `$topic`. Default: `test.topic`. | | `mockSource` | string | no | The value of `$source`. | | `timeoutMs` | integer | no | The evaluation timeout. Default: `100`. QueueBox uses at most `admin.maxTransformTimeoutMs`. | ```bash curl -X POST http://localhost:9090/admin/transform/test \ -H "Authorization: Bearer $ADMIN_TOKEN" \ -H "Content-Type: application/json" \ -d '{ "expression": "{ \"total\": $sum(items.(price * qty)) }", "payload": {"items": [{"price": 10, "qty": 2}, {"price": 5, "qty": 3}]}, "mockTopic": "order.created" }' ``` | Status | Body | Meaning | | ------ | -------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------- | | `200` | `{"success":true,"result":,"context":{"messageId":"","topic":"","attempt":0,"timestamp":""}}` | The expression ran. `context` shows the variables that it saw. | | `400` | `{"success":false,"error":"Invalid request: "}` | The body is not a valid request. | | `400` | `{"success":false,"error":"Invalid expression: "}` | The expression does not compile. | | `400` | `{"success":false,"error":""}` | The evaluation failed or ran out of time. | | `401` | `{"success":false,"error":""}` | The credentials are wrong or absent. | | `413` | `{"success":false,"error":"Request body exceeds bytes"}` | The body is larger than `admin.maxPayloadBytes`. | ### POST /admin/replay [Section titled “POST /admin/replay”](#post-adminreplay) Moves outbox rows back to pending, so the poller delivers them again. It takes only rows that are sent or dead. A row that waits or runs never moves. The move sets `attempt` to `0`, `scheduled_at` to now, and clears `last_error` and the claim columns. The row keeps its `id`. | Field | Type | Meaning | | --------------- | ---------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | | `ids` | array of UUIDs | Only these rows. | | `topic` | string | Only rows with this topic. | | `topics` | array of strings | Only rows with one of these topics. | | `destination` | string | Only rows whose topic the routes send to this destination. The first-match rule of the routes applies. With `topics`, only the topics in both sets. | | `createdAfter` | ISO-8601 instant | Only rows created after this time. | | `createdBefore` | ISO-8601 instant | Only rows created before this time. | Every field is optional, and the fields combine with AND. A request needs at least one field. ```bash curl -X POST http://localhost:9090/admin/replay \ -H "Authorization: Bearer $ADMIN_TOKEN" \ -H "Content-Type: application/json" \ -d '{"destination": "orders-api", "createdAfter": "2026-09-01T00:00:00Z"}' ``` | Status | Body | Meaning | | ------ | ------------------------------------------------- | --------------------------------------------------------------------------- | | `200` | `{"moved":}` | The number of rows that moved. QueueBox also logs the count and the filter. | | `400` | `{"error":"A replay needs at least one filter."}` | The request sets no field. | | `400` | `{"error":"Invalid request: "}` | The body is not a valid request. | | `401` | `{"error":""}` | The credentials are wrong or absent. | | `413` | `{"error":"Request body exceeds bytes"}` | The body is larger than `admin.maxPayloadBytes`. | [Replay dead letters](/how-to/replay-dead-letters/) gives the procedure. # Inbox table > Every column of the inbox table on PostgreSQL and SQL Server, with its type, nullability, default and writer. This page describes every column of the `inbox` table that the bundled migrations create, on PostgreSQL and on SQL Server. QueueBox writes one row for each accepted message of a source. The application reads the table only through a [pull client](/reference/pull-clients/), or for an operational question. `database.inboxTableName` and `database.columnMapping.inbox` rename the table and its columns. See the [configuration](/reference/configuration/#databasecolumnmappinginbox) page. ## Columns [Section titled “Columns”](#columns) | Column | PostgreSQL type | SQL Server type | Null | Default | Written by | | ------------------ | -------------------------- | ------------------ | ---- | ---------------------------------------- | ------------------------------------------ | | `id` | `UUID` | `UNIQUEIDENTIFIER` | no | `gen_random_uuid()` / `NEWID()` | QueueBox, at receipt | | `source` | `VARCHAR(255)` | `NVARCHAR(255)` | no | none | QueueBox, at receipt | | `idempotency_key` | `VARCHAR(255)` | `NVARCHAR(255)` | no | none | QueueBox, at receipt | | `aggregate_id` | `VARCHAR(255)` | `NVARCHAR(255)` | yes | none | QueueBox, at receipt | | `event_type` | `VARCHAR(255)` | `NVARCHAR(255)` | yes | none | QueueBox, at receipt | | `payload` | `JSONB` | `NVARCHAR(MAX)` | no | none | QueueBox, at receipt | | `headers` | `JSONB` | `NVARCHAR(MAX)` | no | `'{}'` | QueueBox, at receipt | | `correlation_id` | `VARCHAR(128)` | `NVARCHAR(128)` | yes | none | QueueBox, at receipt | | `consumption` | `VARCHAR(4)` | `VARCHAR(4)` | no | `'push'` | QueueBox, at receipt | | `state` | `VARCHAR(50)` | `NVARCHAR(50)` | no | `'pending'` | QueueBox, the relay or a pull client | | `created_at` | `TIMESTAMP WITH TIME ZONE` | `DATETIME2` | no | `CURRENT_TIMESTAMP` / `GETUTCDATE()` | The default | | `processed_at` | `TIMESTAMP WITH TIME ZONE` | `DATETIME2` | yes | none | The relay or a pull client | | `scheduled_at` | `TIMESTAMPTZ` | `DATETIME2` | no | `CURRENT_TIMESTAMP` / `SYSUTCDATETIME()` | The default, then a pull client on a retry | | `attempt` | `INT` | `INT` | no | `0` | A pull client on a retry | | `last_error` | `TEXT` | `NVARCHAR(MAX)` | yes | none | A pull client on a retry or a dead letter | | `claimed_at` | `TIMESTAMP WITH TIME ZONE` | `DATETIME2` | yes | none | The relay or a pull client | | `claim_token` | `UUID` | `UNIQUEIDENTIFIER` | yes | none | The relay or a pull client | | `lease_expires_at` | `TIMESTAMPTZ` | `DATETIME2` | yes | none | The relay or a pull client | The pair `(source, idempotency_key)` carries the unique constraint `uq_inbox_source_idempotency`. The constraint is the deduplication. ## Columns set at receipt [Section titled “Columns set at receipt”](#columns-set-at-receipt) ### id [Section titled “id”](#id) The row identifier. The relay sends it on the outbox row as the `x-inbox-id` header. ### source [Section titled “source”](#source) The name of the source in `queuebox.yml`, for example `stripe`. It is the key of the `sources` map, not the path of the HTTP route. ### idempotency\_key [Section titled “idempotency\_key”](#idempotency_key) The deduplication key inside one source. A second message with the same source and key is a duplicate, and QueueBox stores no second row. Two sources can hold the same key, because the source is part of the identity. The [headers](/reference/headers/#headers-that-queuebox-reads) page gives the order in which each kind of source finds the key. When a broker source finds no key, it writes `sha256:` and the hexadecimal SHA-256 digest of the body. ### aggregate\_id [Section titled “aggregate\_id”](#aggregate_id) The unit of order of the inbox. It comes from `aggregateIdPath`, or from a header or the record key on a broker source. At most one row of an aggregate is in flight at a time. The relay copies the value into the outbox `key`. [Ordering](/concepts/ordering/) states the rules for push and pull rows. ### event\_type [Section titled “event\_type”](#event_type) The event type from `eventTypePath`, or from the event type header of a broker source. The topic template reads it as `{{ eventType }}`. ### payload [Section titled “payload”](#payload) The message body as JSON, after the source transform. A broker message whose body is not JSON becomes a dead row with the payload `{"raw": ""}`. ### headers [Section titled “headers”](#headers) A JSON object of the received headers, one string value per name. A repeated name keeps its last value. | Source kind | Content | | ----------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `http` | The request headers, except `Authorization`, `Proxy-Authorization`, `Cookie` and the header that the source `auth` block reads. | | `rabbitmq` | The AMQP headers. A number, a boolean or a timestamp becomes its string form. A nested table or array becomes JSON text. A byte array becomes UTF-8 text, or `base64:` and its Base64 text when it is not valid UTF-8. | | `kafka` | The record headers as UTF-8 text, or `base64:` and the Base64 text when a value is not valid UTF-8. | | `nats` | The message headers. A name with several values keeps its last value. | QueueBox stops at start when the table has no headers column. The error holds the `ALTER TABLE` statement that adds it. ### correlation\_id [Section titled “correlation\_id”](#correlation_id) The identifier that follows the message through every log line. It comes from the `X-Correlation-Id` header, or QueueBox generates a UUID. QueueBox removes control characters and keeps at most 128 characters. ### consumption [Section titled “consumption”](#consumption) `push` or `pull`, from the `consumption` key of the source at receipt. A check constraint allows only these two values. A later change of the configuration does not change a stored row. The relay claims push rows only. ## Columns of the claim [Section titled “Columns of the claim”](#columns-of-the-claim) ### state [Section titled “state”](#state) The position of the row in its life cycle. The meaning of a processed row depends on `consumption`. For a push row, it means that the relay copied the row into the outbox. For a pull row, it means that the application finished its work. [How QueueBox works](/concepts/how-queuebox-works/) lists the states and the transitions between them. ### created\_at and processed\_at [Section titled “created\_at and processed\_at”](#created_at-and-processed_at) `created_at` is the receipt time. The retention age policy of the inbox measures from it. `processed_at` is the time of completion. The inbox has no `updated_at` column. ### scheduled\_at and attempt [Section titled “scheduled\_at and attempt”](#scheduled_at-and-attempt) A pull claim takes a row only when `scheduled_at` has passed. A pull retry moves `scheduled_at` forward and raises `attempt` by one. The relay does not use these two columns. ### last\_error [Section titled “last\_error”](#last_error) The reason that a pull client gives on a retry or a dead letter. The pull clients remove secret values from the text before they write it. ### claimed\_at, claim\_token and lease\_expires\_at [Section titled “claimed\_at, claim\_token and lease\_expires\_at”](#claimed_at-claim_token-and-lease_expires_at) A claim sets all three: the claim time, a new random token and the end of the lease. Every completion, retry, renewal and dead letter matches the token and a lease that has not passed. An update that matches no row means that the claim was lost. A pull completion, retry or dead letter clears `claim_token` and `lease_expires_at`. [Claims and leases](/concepts/claims-and-leases/) explains the model. ## Indexes [Section titled “Indexes”](#indexes) * PostgreSQL | Index | Columns | Filter | | ------------------------------- | -------------------------------------------- | -------------------- | | Primary key | `id` | none | | `uq_inbox_source_idempotency` | `source`, `idempotency_key` | unique | | `idx_inbox_pending` | `state` | pending rows | | `idx_inbox_source` | `source` | none | | `idx_inbox_aggregate_state` | `aggregate_id`, `state` | none | | `idx_inbox_processing_claimed` | `claimed_at` | processing rows | | `idx_inbox_state_created` | `state`, `created_at` | none | | `idx_inbox_consumption_pending` | `consumption`, `state`, `scheduled_at` | none | | `idx_inbox_pull_pending` | `source`, `scheduled_at`, `created_at`, `id` | pending pull rows | | `idx_inbox_pull_busy` | `source`, `aggregate_id`, `lease_expires_at` | processing pull rows | * SQL Server | Index | Columns | Filter | | ------------------------------- | -------------------------------------------- | -------------------- | | Primary key | `id` | none | | `uq_inbox_source_idempotency` | `source`, `idempotency_key` | unique | | `idx_inbox_state` | `state` | none | | `idx_inbox_state_created` | `state`, `created_at` | none | | `idx_inbox_aggregate_state` | `aggregate_id`, `state` | none | | `idx_inbox_processing_claimed` | `claimed_at` | processing rows | | `idx_inbox_consumption_pending` | `consumption`, `state`, `scheduled_at` | none | | `idx_inbox_pull_pending` | `source`, `scheduled_at`, `created_at`, `id` | pending pull rows | | `idx_inbox_pull_busy` | `source`, `aggregate_id`, `lease_expires_at` | processing pull rows | ## Example query [Section titled “Example query”](#example-query) Read the table for an operational question, for example to check that a webhook arrived. * PostgreSQL ```sql SELECT id, source, idempotency_key, event_type, state, created_at, processed_at FROM inbox WHERE source = 'stripe' ORDER BY created_at DESC LIMIT 20; ``` * SQL Server ```sql SELECT TOP 20 id, source, idempotency_key, event_type, state, created_at, processed_at FROM inbox WHERE source = N'stripe' ORDER BY created_at DESC; ``` # Metrics > Every metric that the Prometheus scrape of QueueBox exposes, with its type, tags and meaning. This page lists every metric that `GET /metrics` exposes. The body uses the Prometheus text format. The route listens on `server.managementPort` when the configuration sets one. [Monitoring](/operations/monitoring/) gives the alerts to build on these metrics. The test `MetricsDocTest` compares this page with a live scrape. Every name in the tables must appear in the scrape. Every name in the scrape must appear in a table, or match a prefix of the allowlist at the end of this page. ## How the exporter changes a name [Section titled “How the exporter changes a name”](#how-the-exporter-changes-a-name) The Prometheus exporter renames two kinds of metric: * It removes the `_info` suffix. The registered gauge `queuebox_info` appears as `queuebox`. * It adds a second family for a timer. A timer named `x_seconds` also exposes the gauge `x_seconds_max`. The gauge holds the largest value of the current decay window. A counter keeps its `_total` suffix in the `# TYPE` line and in the sample line. ## QueueBox metrics [Section titled “QueueBox metrics”](#queuebox-metrics) Each tag value comes from the configuration or from a fixed set of values, so the number of label sets stays bounded. | Metric | Type | Tags | Meaning | | ------------------------------------------------- | ------- | ---------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `queuebox` | gauge | `version` | The build information. The value is always 1. The registered name is `queuebox_info`. | | `queuebox_uptime_seconds` | gauge | none | The seconds since the process started. | | `queuebox_outbox_messages_total` | counter | `status`: sent, failed or dead | The outbox delivery outcomes. A failed outcome is a failed delivery that QueueBox retries. | | `queuebox_outbox_messages_pending` | gauge | none | The outbox messages that wait for a publish. | | `queuebox_outbox_messages_reclaimed_total` | counter | none | The outbox messages that returned to pending after a stale claim. | | `queuebox_outbox_oldest_pending_age_seconds` | gauge | none | The age in seconds of the oldest outbox row in state `pending`. The value is zero when no row is pending. A count of pending rows cannot separate a busy relay from a dead one, so this is the metric to alert on. The poller refreshes the value at most once per `outbox.pendingGaugeIntervalMs`, so the value can be up to that interval old. The gauge does not run a query on each scrape. | | `queuebox_claims_lost_total` | counter | `component` | The terminal writes that lost the claim. Another replica owned the message. | | `queuebox_outbox_process_errors_total` | counter | none | The errors that stopped the processing of one outbox message. | | `queuebox_outbox_processing_duration_seconds` | summary | none | The time to process one outbox message. The summary carries the 50th, 95th and 99th percentile. | | `queuebox_outbox_processing_duration_seconds_max` | gauge | none | The largest processing time of the current window. | | `queuebox_outbox_publish_duration_seconds` | summary | `destination_type` | The time to publish one message to a destination type, for example `http`. | | `queuebox_outbox_publish_duration_seconds_max` | gauge | `destination_type` | The largest publish time of the current window. | | `queuebox_outbox_destination_messages_total` | counter | `destination`, `outcome` = `success` or `failure` | The outbox messages per destination and outcome. | | `queuebox_outbox_queue_depth` | gauge | `destination` | The messages that wait for a publish to one destination. | | `queuebox_http_publish_responses_total` | counter | `status_class` = `1xx`, `2xx`, `3xx`, `4xx`, `5xx` or `other` | The HTTP publish responses per status class. A raw status code is never a label. | | `queuebox_transform_failures_total` | counter | `strategy` | The transform failures per error strategy. | | `queuebox_inbox_messages_total` | counter | `status` = `new`, `forwarded` or `duplicate` | The inbox messages per status. | | `queuebox_inbox_relay_errors_total` | counter | none | The errors of the inbox relay. | | `queuebox_inbox_rejections_total` | counter | `reason` = `extraction_failed`, `transform_failed` or `storage_failed` | The inbox messages that QueueBox rejected, per reason. | | `queuebox_inbox_filtered_total` | counter | `source` | The inbox messages that the header filter of a source dropped. QueueBox stores no row for them. The counter appears after the first drop of a source. | | `queuebox_inbox_oldest_pending_age_seconds` | gauge | none | The age in seconds of the oldest inbox row in state `pending`. The value is zero when no row is pending. A count of pending rows cannot separate a busy relay from a dead one, so this is the metric to alert on. The relay refreshes the value at most once per `inbox.relay.pendingGaugeIntervalMs`, so the value can be up to that interval old. The gauge does not run a query on each scrape. | | `queuebox_cleanup_messages_deleted_total` | counter | `table` | The rows that the retention cleanup deleted, per table. | | `queuebox_cleanup_duration_seconds` | summary | `table` | The time of one cleanup run, per table. | | `queuebox_cleanup_duration_seconds_max` | gauge | `table` | The longest cleanup run of the current window. | | `queuebox_cleanup_last_run_timestamp` | gauge | `table` | The Unix time in seconds of the last cleanup run. | A metric with a tag registers on its first use. A destination that never received a message has no sample until its first publish. ## HikariCP pool metrics [Section titled “HikariCP pool metrics”](#hikaricp-pool-metrics) `DatabaseFactory` gives the meter registry to HikariCP through `MicrometerMetricsTrackerFactory`. Every metric below carries the tag `pool`, which holds the pool name. | Metric | Type | Tags | Meaning | | ------------------------------------------- | ------- | ------ | ------------------------------------------------------------ | | `hikaricp_connections` | gauge | `pool` | The connections in the pool, both idle and active. | | `hikaricp_connections_active` | gauge | `pool` | The connections that a caller holds. | | `hikaricp_connections_idle` | gauge | `pool` | The connections that are free. | | `hikaricp_connections_pending` | gauge | `pool` | The threads that wait for a connection. | | `hikaricp_connections_min` | gauge | `pool` | The minimum idle connection count of the pool configuration. | | `hikaricp_connections_max` | gauge | `pool` | The maximum pool size of the pool configuration. | | `hikaricp_connections_timeout_total` | counter | `pool` | The connection requests that timed out. | | `hikaricp_connections_acquire_seconds` | summary | `pool` | The time to acquire a connection from the pool. | | `hikaricp_connections_acquire_seconds_max` | gauge | `pool` | The longest acquire time of the current window. | | `hikaricp_connections_creation_seconds` | summary | `pool` | The time to create a physical connection. | | `hikaricp_connections_creation_seconds_max` | gauge | `pool` | The longest creation time of the current window. | | `hikaricp_connections_usage_seconds` | summary | `pool` | The time a caller held a connection. | | `hikaricp_connections_usage_seconds_max` | gauge | `pool` | The longest usage time of the current window. | ## JVM metrics [Section titled “JVM metrics”](#jvm-metrics) QueueBox binds no JVM metrics. No `jvm_`, `process_` or `system_` family appears in the scrape, because the application registers no Micrometer JVM binder. Read the heap, the thread count and the garbage collection from another exporter. Do not add a JVM family to this page before a binder exists. `MetricsDocTest` fails on a documented metric that the scrape does not carry. ## Allowlist [Section titled “Allowlist”](#allowlist) The allowlist names the prefixes that this page does not list one by one. It is empty. Every metric of the scrape has its own row above. # Outbox table > Every column of the outbox table on PostgreSQL and SQL Server, with its type, nullability, default and writer. This page describes every column of the `outbox` table that the bundled migrations create, on PostgreSQL and on SQL Server. The migrations live in [`postgres/src/main/resources/db/postgresql`](https://github.com/alternayte/queuebox/blob/main/postgres/src/main/resources/db/postgresql) and [`sqlserver/src/main/resources/db/sqlserver`](https://github.com/alternayte/queuebox/blob/main/sqlserver/src/main/resources/db/sqlserver). `database.outboxTableName` and `database.columnMapping.outbox` rename the table and its columns. See the [configuration](/reference/configuration/#databasecolumnmappingoutbox) page. ## Columns [Section titled “Columns”](#columns) | Column | PostgreSQL type | SQL Server type | Null | Default | Written by | | ------------------ | -------------------------- | ------------------ | ---- | ----------------------------------------------------------------------- | ----------------------------------------------------------------------- | | `id` | `UUID` | `UNIQUEIDENTIFIER` | no | `gen_random_uuid()` / `NEWID()` | The application, or the default | | `topic` | `VARCHAR(255)` | `NVARCHAR(255)` | no | none | The application. Required. | | `key` | `VARCHAR(255)` | `NVARCHAR(255)` | yes | none | The application. Optional. | | `payload` | `JSONB` | `NVARCHAR(MAX)` | no | none | The application. Required. | | `headers` | `JSONB` | `NVARCHAR(MAX)` | no | `'{}'` | The application. Optional. | | `aggregate_type` | `VARCHAR(255)` | `NVARCHAR(255)` | yes | none | The application. Optional. | | `state` | `VARCHAR(50)` | `NVARCHAR(50)` | no | `'pending'` | QueueBox | | `attempt` | `INTEGER` | `INT` | no | `0` | QueueBox | | `max_attempts` | `INTEGER` | `INT` | no | `5` | The application, or the default. The relay writes `outbox.maxAttempts`. | | `scheduled_at` | `TIMESTAMP WITH TIME ZONE` | `DATETIME2` | no | `CURRENT_TIMESTAMP` / `GETUTCDATE()` | The application, or the default. QueueBox on a retry. | | `created_at` | `TIMESTAMP WITH TIME ZONE` | `DATETIME2` | no | `CURRENT_TIMESTAMP` / `GETUTCDATE()` | The default | | `updated_at` | `TIMESTAMP WITH TIME ZONE` | `DATETIME2` | no | `CURRENT_TIMESTAMP` / `GETUTCDATE()` | QueueBox | | `claimed_at` | `TIMESTAMP WITH TIME ZONE` | `DATETIME2` | yes | none | QueueBox | | `claim_token` | `UUID` | `UNIQUEIDENTIFIER` | yes | none | QueueBox | | `lease_expires_at` | `TIMESTAMPTZ` | `DATETIME2` | yes | none | QueueBox | | `last_error` | `TEXT` | `NVARCHAR(MAX)` | yes | none | QueueBox | | `sequence` | `BIGINT` | `BIGINT` | no | `nextval('outbox_sequence_seq')` / `NEXT VALUE FOR outbox_sequence_seq` | The default. Do not write it. | An insert needs two columns only: `topic` and `payload`. Every other column has a default or accepts null. On SQL Server, `key` is a reserved word, so an insert writes it as `[key]`. ## Columns the application writes [Section titled “Columns the application writes”](#columns-the-application-writes) ### id [Section titled “id”](#id) The row identifier. QueueBox sends it as `X-Message-Id` on an HTTP delivery, and as the message identifier on a broker. It stays the same on every retry of the row. A replay of a row keeps the identifier. ### topic [Section titled “topic”](#topic) The routing input. QueueBox matches it against the `topicPattern` of each route, in list order, and the first match wins. A row whose topic matches no route goes dead. At most 255 characters. ### key [Section titled “key”](#key) The unit of order. The rows of one non-empty key reach the destination in insert order, at any concurrency and with any number of replicas. A null or empty key takes part in no order. The relay writes the inbox `aggregate_id` here. [Ordering](/concepts/ordering/) states the rules. ### payload [Section titled “payload”](#payload) The message body, as one JSON document. An HTTP destination receives it as the request body. Write a JSON object or array, not a JSON string of JSON text. ### headers [Section titled “headers”](#headers) A JSON object of string values. A destination sends each entry as a message header. The column is `NOT NULL`. Omit the column, or write a JSON object. An explicit `NULL` fails the insert. The [headers](/reference/headers/) page lists the headers that QueueBox adds. ### aggregate\_type [Section titled “aggregate\_type”](#aggregate_type) The kind of business entity, for example `order`. A RabbitMQ, Kafka or NATS destination can build its address from it, through `{{ aggregateType }}` or `exchangeFrom: aggregate_type`. ### max\_attempts [Section titled “max\_attempts”](#max_attempts) The dead-letter ceiling of this row. QueueBox reads this column, not the configuration, when a delivery fails: * A failed delivery with `attempt` below `max_attempts` schedules a retry and raises `attempt` by one. * A failed delivery with `attempt` equal to `max_attempts` marks the row dead. The ceiling comes from the first of these that sets it: 1. The insert of the application. 2. `inbox.relay.maxAttempts`, or else `outbox.maxAttempts`, for a row that the relay creates. 3. The column default `5`. ### scheduled\_at [Section titled “scheduled\_at”](#scheduled_at) The earliest time of the next delivery. The poller claims a row only when `scheduled_at` has passed. Set it in the future to delay a message. A retry moves it forward by the backoff. ## Columns that QueueBox writes [Section titled “Columns that QueueBox writes”](#columns-that-queuebox-writes) Do not write these columns. A row that the application creates with another state or a claim can stall, or can deliver twice. ### state [Section titled “state”](#state) The position of the row in its life cycle. An insert takes the default. [How QueueBox works](/concepts/how-queuebox-works/) lists the states and the transitions between them. ### attempt [Section titled “attempt”](#attempt) The number of failed deliveries. It is `0` on the first delivery. QueueBox sends it as `X-Attempt`, and a transform reads it as `$attempt`. A replay resets it to `0`. ### created\_at and updated\_at [Section titled “created\_at and updated\_at”](#created_at-and-updated_at) `created_at` is the insert time. `updated_at` is the time of the last state change. The retention age policy of the outbox measures from `updated_at`. ### claimed\_at, claim\_token and lease\_expires\_at [Section titled “claimed\_at, claim\_token and lease\_expires\_at”](#claimed_at-claim_token-and-lease_expires_at) A claim sets all three: the claim time, a new random token and the end of the lease. Every terminal write matches the token, so a replica that lost its claim cannot complete the row. The lease length is `outbox.claimTimeoutMs`, and the poller renews the lease while a publish runs. A row whose lease has passed returns to the claimable state. [Claims and leases](/concepts/claims-and-leases/) explains the model. ### last\_error [Section titled “last\_error”](#last_error) The reason of the last failed delivery. QueueBox removes secret values and cuts the text to `http.maxErrorBodyBytes` before it writes the column. A replay clears it. ### sequence [Section titled “sequence”](#sequence) The insert order. The database fills it from the sequence `outbox_sequence_seq`. The claim orders the rows of one key by `sequence`, not by `created_at`, because rows of one transaction share a `created_at`. Migration V11 adds the column and numbers the rows that exist, in `created_at` order. ## Indexes [Section titled “Indexes”](#indexes) * PostgreSQL | Index | Columns | Filter | | ------------------------------- | ----------------------- | --------------------------- | | Primary key | `id` | none | | `idx_outbox_pending_scheduled` | `state`, `scheduled_at` | pending rows | | `idx_outbox_topic` | `topic` | none | | `idx_outbox_processing_claimed` | `claimed_at` | processing rows | | `idx_outbox_key_sequence` | `key`, `sequence` | pending and processing rows | * SQL Server | Index | Columns | Filter | | ------------------------------- | ----------------------- | --------------------------- | | Primary key | `id` | none | | `idx_outbox_pending_scheduled` | `state`, `scheduled_at` | pending rows | | `idx_outbox_state_updated` | `state`, `updated_at` | none | | `idx_outbox_processing_claimed` | `claimed_at` | processing rows | | `idx_outbox_key_sequence` | `[key]`, `sequence` | pending and processing rows | ## Example insert [Section titled “Example insert”](#example-insert) The insert runs in the transaction of the business write. [Write outbox rows](/how-to/write-outbox-rows/) gives the full procedure. * PostgreSQL ```sql INSERT INTO outbox (topic, key, payload, headers, aggregate_type) VALUES ( 'order.created', 'cust-42', '{"orderId":"11111111-1111-1111-1111-111111111111","amount":99.99}'::jsonb, '{"X-Tenant":"acme"}'::jsonb, 'order' ); ``` * SQL Server ```sql INSERT INTO outbox (topic, [key], payload, headers, aggregate_type) VALUES ( N'order.created', N'cust-42', N'{"orderId":"11111111-1111-1111-1111-111111111111","amount":99.99}', N'{"X-Tenant":"acme"}', N'order' ); ``` # Pull clients > The Go, TypeScript and C# libraries that claim and complete pull inbox rows, with their install command, API surface and settings. This page describes the three pull-inbox libraries: Go, TypeScript and C#. Each library claims the rows of one `consumption: pull` source, runs your handler in a transaction, and completes, retries or dead-letters the row. The README of each library holds the full text: [Go](https://github.com/alternayte/queuebox/blob/main/clients/go/README.md), [TypeScript](https://github.com/alternayte/queuebox/blob/main/clients/typescript/README.md) and [C#](https://github.com/alternayte/queuebox/blob/main/clients/csharp/README.md). [Consume the inbox](/how-to/consume-the-inbox/) shows a worker from start to end. ## What every library does [Section titled “What every library does”](#what-every-library-does) | Behaviour | Rule | | ------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Claim | One transaction holds the claim and nothing else, and it commits before a handler starts. | | Handler transaction | The handler receives an open transaction. The library runs the completion in that transaction, then commits. | | Lost lease | A completion that updates no row means that another worker owns the row. The library rolls the handler writes back. | | Renewal | The library renews the lease every third of the lease duration. | | Failure | A handler error rolls the transaction back. The retry policy then retries the row later or dead-letters it. | | Shutdown | A stop ends the claims at once. A handler that runs past the grace period is cancelled. The library completes nothing for it and spends no attempt, and the lease expires. | | Redaction | Every log line and every `last_error` value passes through the redaction that QueueBox uses. | | Schema | The library writes no schema. It needs the V10 schema or later, which adds the inbox `headers` column. | | Claim token | The handler never sees the token. | Rules for a handler, in every language: 1. Write every change through the transaction that the handler receives. 2. Do not commit and do not roll back. The library owns both. 3. Fail the row with an error or an exception. 4. Honour the cancellation. It fires when the lease is lost and when a shutdown runs out of grace. 5. Deduplicate external work on the source and the idempotency key. A transaction cannot roll back a call to another system. ## Message fields [Section titled “Message fields”](#message-fields) | Field | Meaning | | --------------- | --------------------------------------------------------------------------------------- | | id | The inbox row identifier. | | source | The source name. | | idempotency key | The deduplication key. The identity of the message is the source and this key together. | | aggregate id | Nullable. | | event type | Nullable. | | payload | The JSON body. | | headers | The stored headers, one string value per name. Empty when there are none. | | attempt | `0` on the first delivery. | | correlation id | Nullable. For logs. | ## Go [Section titled “Go”](#go) | | | | ----------- | -------------------------------------------------------------------------------------------------------------- | | Module | `github.com/alternayte/queuebox/clients/go`, package `queuebox` | | Go | 1.24 or later | | Drivers | `database/sql` with `pgx` for PostgreSQL or `go-mssqldb` for SQL Server. The application registers the driver. | | Release tag | `clients/go/vX.Y.Z` | ```bash go get github.com/alternayte/queuebox/clients/go ``` ```go worker, err := queuebox.NewInboxWorker(db, queuebox.Options{Source: "orders", BatchSize: 10, LeaseMS: 30_000}) if err != nil { log.Fatal(err) } err = worker.Run(ctx, func(ctx context.Context, message queuebox.Message, tx *sql.Tx) error { var order struct { ID string `json:"id"` Total int `json:"total"` } if err := message.UnmarshalPayload(&order); err != nil { return err } _, err := tx.ExecContext(ctx, "INSERT INTO orders (id, total) VALUES ($1, $2)", order.ID, order.Total) return err }) ``` ### API [Section titled “API”](#api) | Name | Kind | Purpose | | ------------------------------------------------------------------- | --------- | --------------------------------------------------------------------------------------------------------------- | | `NewInboxWorker(db *sql.DB, options Options) (*InboxWorker, error)` | function | Builds a worker. It returns an error for a setting that cannot work. | | `(*InboxWorker).Run(ctx, handler) error` | method | Claims and handles rows until `ctx` is cancelled. | | `Handler` | type | `func(ctx context.Context, message Message, tx *sql.Tx) error` | | `Message` | struct | `ID`, `Source`, `IdempotencyKey`, `AggregateID`, `EventType`, `Payload`, `Attempt`, `CorrelationID`, `Headers`. | | `(Message).UnmarshalPayload(target any) error` | method | Parses the payload into `target`. | | `RetryPolicy` | interface | `Decide(message Message, failure error) FailureAction` | | `RetryPolicyFunc` | type | A function that satisfies `RetryPolicy`. | | `NewDefaultRetryPolicy(maxAttempts, baseDelay, maxDelay, jitter)` | function | The exponential policy with jitter. | | `RetryAfter(delay)`, `DeadLetter()` | functions | The two `FailureAction` values. | | `DefaultSchema() Schema` | function | The default table and column names, to change for a mapped schema. | | `DialectPostgreSQL`, `DialectSQLServer` | constants | The values of `Options.Dialect`. | | `Logger` | interface | `Info`, `Warn` and `Error`, each with a message and a map of fields. | | `Sanitize`, `SanitizeError` | functions | The redaction that the library applies. | ### Options [Section titled “Options”](#options) | Field | Default | Meaning | | ---------------- | ------------------------------------------------------- | ------------------------------------------------- | | `Source` | required | The source whose rows this worker takes. | | `BatchSize` | `10` | The largest number of rows in one claim. | | `LeaseMS` | `30000` | The lease duration in milliseconds. | | `MaxConcurrency` | `1` | The largest number of handlers at one time. | | `PollInterval` | 1 second | The wait after a claim that returned nothing. | | `ShutdownGrace` | 30 seconds | How long a stop waits for running handlers. | | `Dialect` | `DialectPostgreSQL` | The database dialect. | | `Schema` | the default names | A `*Schema` for a mapped table or mapped columns. | | `RetryPolicy` | 5 attempts, 1 second base, 5 minute ceiling, 0.2 jitter | What happens to a row whose handler failed. | | `Logger` | none | Without a logger, the library prints nothing. | On SQL Server, give each claim call a context deadline of at least 30 seconds. ## TypeScript [Section titled “TypeScript”](#typescript) | | | | ----------- | ------------------------------------------------------------------------------------------------- | | Package | `@alternayte/queuebox-inbox` on npm, ESM and CJS | | Runtime | Node 22 or later, Bun, or Deno | | Drivers | `pg` for PostgreSQL or `mssql` for SQL Server, through an adapter. The package imports no driver. | | Release tag | `typescript-vX.Y.Z` | ```bash npm install @alternayte/queuebox-inbox ``` ```ts import pg from "pg"; import { InboxWorker, fromPg } from "@alternayte/queuebox-inbox"; const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL }); const worker = new InboxWorker(fromPg(pool), { source: "orders", batchSize: 10, leaseMs: 30_000 }); await worker.run(async (message, tx) => { const payload = message.payload as { id: string; total: number }; await tx.query("INSERT INTO orders (id, total) VALUES ($1, $2)", [payload.id, payload.total]); }); ``` ### API [Section titled “API”](#api-1) | Name | Kind | Purpose | | ----------------------------------------------- | ------------------- | --------------------------------------------------------------------------------------------------------------- | | `new InboxWorker(connection, options)` | class | Builds a worker. | | `worker.run(handler, signal?)` | method | Claims and handles rows until the signal aborts. | | `InboxHandler` | type | `(message, tx, signal) => Promise` | | `fromPg(pool)` | function | The adapter for a `pg` pool. | | `fromMssql(mssql, pool)` | function | The adapter for an `mssql` pool. It takes the `mssql` module itself. | | `InboxMessage` | type | `id`, `source`, `idempotencyKey`, `aggregateId`, `eventType`, `payload`, `headers`, `attempt`, `correlationId`. | | `InboxRetryPolicy` | type | `(message, failure) => InboxFailureAction` | | `defaultRetryPolicy(options?)` | function | The exponential policy with jitter. | | `retryAfter(ms)`, `deadLetter()` | functions | The two `InboxFailureAction` values. | | `defaultSchema` | value | The default table and column names. | | `inboxSql` | function | The SQL statements for a dialect and a schema. | | `sanitize`, `sanitizeError`, `MAX_ERROR_LENGTH` | functions and value | The redaction that the library applies. | | `resolveOptions` | function | The options with their defaults filled in. | ### Options [Section titled “Options”](#options-1) | Field | Default | Meaning | | ----------------- | ---------------------- | -------------------------------------------------------- | | `source` | required | The source whose rows this worker takes. | | `batchSize` | `10` | The largest number of rows in one claim. | | `leaseMs` | `30000` | The lease duration in milliseconds. | | `maxConcurrency` | `1` | The largest number of handlers at one time. | | `pollIntervalMs` | `1000` | The wait after a claim that returned nothing. | | `shutdownGraceMs` | `30000` | How long a stop waits for running handlers. | | `dialect` | `"postgresql"` | `"postgresql"` or `"sqlserver"`. | | `schema` | `defaultSchema` | The names for a mapped table or mapped columns. | | `retryPolicy` | `defaultRetryPolicy()` | 5 attempts, 1000 ms base, 300000 ms ceiling, 0.2 jitter. | | `logger` | none | Without a logger, the library prints nothing. | On SQL Server, set the `mssql` request timeout to at least 30 seconds. Its default of 15 seconds is too short. ## C\# [Section titled “C#”](#c) | | | | ----------- | ------------------------------------------------------------------------------------------------------------------------------------- | | Packages | `QueueBox.Inbox`, and `QueueBox.Inbox.DependencyInjection` for a hosted service, on NuGet | | Framework | .NET 8.0 or later | | Drivers | `Npgsql` for PostgreSQL or `Microsoft.Data.SqlClient` for SQL Server, through `System.Data.Common`. The package depends on no driver. | | Release tag | `csharp-vX.Y.Z` | ```bash dotnet add package QueueBox.Inbox ``` ```csharp await using var dataSource = NpgsqlDataSource.Create(connectionString); var worker = new InboxWorker( InboxConnections.From(dataSource), new InboxOptions { Source = "orders", BatchSize = 10, LeaseMs = 30_000 }); await worker.RunAsync(async (message, transaction, cancellationToken) => { await using var command = transaction.CreateCommand(); command.CommandText = "INSERT INTO orders (id, total) VALUES (@id, @total)"; command .WithParameter("@id", message.Payload.GetProperty("id").GetString()) .WithParameter("@total", message.Payload.GetProperty("total").GetInt32()); await command.ExecuteNonQueryAsync(cancellationToken); }); ``` ### API [Section titled “API”](#api-2) | Name | Kind | Purpose | | ------------------------------------------------------------------------- | ----------------- | -------------------------------------------------------------------------------------------------------------------------------- | | `InboxWorker(connections, options, logger?)` | class | Builds a worker. `logger` is an `ILogger`. | | `InboxWorker.RunAsync(handler, cancellationToken)` | method | Claims and handles rows until the token fires. | | `InboxHandler` | delegate | `(InboxMessage message, DbTransaction transaction, CancellationToken token) => Task` | | `InboxConnections.From(DbDataSource)` | method | A connection source for a data source, for example Npgsql. | | `InboxConnections.From(DbProviderFactory, string)` | method | A connection source for a factory and a connection string, for example `SqlClientFactory.Instance`. | | `InboxConnections.From(Func>)` | method | A connection source for your own open function. | | `IInboxConnectionSource` | interface | The connection source that a worker uses. | | `InboxMessage` | class | `Id`, `Source`, `IdempotencyKey`, `AggregateId`, `EventType`, `Payload` as `JsonElement`, `Attempt`, `CorrelationId`, `Headers`. | | `IInboxRetryPolicy` | interface | `InboxFailureAction Decide(InboxMessage message, Exception failure)` | | `DefaultRetryPolicy(maxAttempts, baseDelay, maxDelay, jitter)` | class | The exponential policy with jitter. | | `InboxFailureAction.Retry(TimeSpan)`, `InboxFailureAction.DeadLetter()` | methods | The two failure actions. | | `InboxSchema.Default` | record | The default names. Change them with a `with` expression. | | `SqlDialect` | enum | `PostgreSql` or `SqlServer`. | | `transaction.CreateCommand()`, `command.WithParameter(name, value)` | extension methods | Helpers for a command on the handler transaction. | | `services.AddQueueBoxInbox(name, options, handler, connections?)` | extension method | `QueueBox.Inbox.DependencyInjection`. Registers a named worker as a hosted service. | | `InboxDbContextFactory.CreateOn(transaction, build)` | method | `QueueBox.Inbox.DependencyInjection`. Builds an Entity Framework Core context on the handler transaction. | ### Options [Section titled “Options”](#options-2) | Property | Default | Meaning | | ---------------- | --------------------- | -------------------------------------------------------- | | `Source` | required | The source whose rows this worker takes. | | `BatchSize` | `10` | The largest number of rows in one claim. | | `LeaseMs` | `30000` | The lease duration in milliseconds. | | `MaxConcurrency` | `1` | The largest number of handlers at one time. | | `PollInterval` | 1 second | The wait after a claim that returned nothing. | | `ShutdownGrace` | 30 seconds | How long a stop waits for running handlers. | | `Dialect` | `PostgreSql` | `PostgreSql` or `SqlServer`. | | `Schema` | `InboxSchema.Default` | The names for a mapped table or mapped columns. | | `RetryPolicy` | `DefaultRetryPolicy` | 5 attempts, 1 second base, 5 minute ceiling, 0.2 jitter. | On SQL Server, keep the `Microsoft.Data.SqlClient` command timeout at 30 seconds or more. The default is 30 seconds. Entity Framework Core opens its own connection by default. Build the context through `InboxDbContextFactory.CreateOn`, so the application write and the completion share one transaction. ## The retry policy [Section titled “The retry policy”](#the-retry-policy) The default policy of each library retries while `attempt` is below the ceiling of 5, and dead-letters after that. The delay is the base delay times `2^attempt`, at most the ceiling delay, with a random jitter of 20 percent. The row starts at attempt `0`, so the default allows six deliveries. Write your own policy when an error must never retry, for example a payload that does not parse. ## The SQL contract [Section titled “The SQL contract”](#the-sql-contract) A language without a library can use the SQL statements in [`examples/pull/sql`](https://github.com/alternayte/queuebox/tree/main/examples/pull/sql) directly. The [pull example README](https://github.com/alternayte/queuebox/blob/main/examples/pull/README.md) holds the full rules. | File | Parameters | Effect | | -------------- | ----------------------------------------------- | ---------------------------------------------------------------------------------------------------- | | `claim.sql` | `:source`, `:batch`, `:lease_ms`, `:cand_limit` | Claims up to `batch` rows of the source, one per aggregate. Returns the rows with their claim token. | | `renew.sql` | `:id`, `:token`, `:lease_ms` | Extends the lease. | | `complete.sql` | `:id`, `:token` | Marks the row processed. Run it in the transaction of the business writes. | | `retry.sql` | `:id`, `:token`, `:delay_ms`, `:error` | Returns the row to pending after the delay, and raises `attempt` by one. | | `dead.sql` | `:id`, `:token`, `:error` | Marks the row dead. | * Bind every value as a parameter. Do not build the SQL text from values. * Compute `cand_limit` as `LEAST(GREATEST(3 * batch, 50), 500)`, and bind it on every call. * Renew, complete, retry and dead-letter must each update exactly one row. Zero rows means that the claim was lost. Stop the work. * Remove secrets from `error` and cut its length before you bind it. On SQL Server, the claim takes an application lock per source with a 10 second lock timeout. A lock failure raises Msg 51000. Treat it as a transient error, and wait before the next claim. Set the driver timeout to at least 30 seconds, so the server raises Msg 51000 before the driver aborts the call. An aborted call leaves the lock held until the connection resets. A SQL Server claim needs READ COMMITTED isolation. It also works with READ\_COMMITTED\_SNAPSHOT ON. # Transforms and templates > The JSONata variables of each transform stage, the error strategies, and the placeholders of every QueueBox template. This page lists the variables that a JSONata transform reads at each stage, the error strategies, and the placeholders of the topic, address and routing key templates. [Transform payloads](/how-to/transform-payloads/) shows how to write a transform. ## Stages [Section titled “Stages”](#stages) QueueBox runs a [JSONata](https://jsonata.org/) expression at three stages. The [configuration](/reference/configuration/#transform) page lists the keys of a `transform` block. | Stage | Configured on | Runs | Input | | ----------- | ------------------------------- | ------------------------------------------------------------------------------- | ----------------------------- | | Source | `sources..transform` | Once, when the message arrives, before the store and before the duplicate check | The received body | | Route | `routes[i].transform` | On every delivery attempt, before the destination transform | The outbox `payload` | | Destination | `destinations..transform` | On every delivery attempt, after the route transform | The output of the route stage | The source stage changes the stored inbox payload. The route and destination stages change only the delivered body. The outbox row keeps its payload. QueueBox compiles every expression at start. An expression that does not compile stops the start, and the error names the key path. ## Variables [Section titled “Variables”](#variables) A variable starts with `$`. `$` alone is the input document. | Variable | Source stage | Route and destination stages | | ----------------- | ----------------------------------------------- | ----------------------------------------------------------------- | | `$messageId` | The `id` of the new inbox row | The `id` of the outbox row | | `$topic` | The extracted event type, or an empty string | The `topic` of the outbox row | | `$attempt` | `0` | The `attempt` of the outbox row. It is `0` on the first delivery. | | `$timestamp` | The receipt time, as ISO-8601 text | The `created_at` of the outbox row, as ISO-8601 text | | `$source` | The source name | not bound | | `$headers` | The received headers, one string value per name | not bound | | `$idempotencyKey` | The extracted idempotency key, or `null` | not bound | | `$eventType` | The extracted event type, or `null` | not bound | QueueBox 0.4.0 and earlier do not bind `$idempotencyKey` and `$eventType`. On those versions, read the event type from `$topic`. A header name with a `-` needs quotes: `$headers."x-tenant"`. A header name matches in its exact letter case, so use the case that the inbox row shows. An HTTP source omits the credential headers from `$headers`. The [inbox table](/reference/inbox-table/#headers) page lists them. `POST /admin/transform/test` binds `$messageId` to a random UUID, `$topic` to `mockTopic`, `$attempt` to `0`, `$timestamp` to the current time, and `$source` to `mockSource`. It binds no `$headers`. ## Limits [Section titled “Limits”](#limits) | Key | Default | Effect | | ----------- | ------- | ----------------------------------------- | | `timeoutMs` | `100` | An evaluation that runs longer fails. | | `maxDepth` | `100` | An evaluation that recurses deeper fails. | A failure of either limit is a transform error. The error strategy of the block decides what happens next. ## Error strategies [Section titled “Error strategies”](#error-strategies) `onError` takes `Fail`, `Skip` or `Dead`, in this letter case. The default is `Fail`. | Strategy | Route or destination stage | Source stage, HTTP | Source stage, RabbitMQ, Kafka or NATS | | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------- | -------------------------------------------------------------------------------------- | | `Fail` | The delivery fails. The row retries with backoff and counts an attempt. | `422`. QueueBox stores no row. | QueueBox stores the original payload as a dead inbox row and acknowledges the message. | | `Skip` | The stage passes its input on unchanged. The route stage passes the outbox payload. The destination stage passes the output of the route stage. | QueueBox stores the original body. | QueueBox stores the original body. | | `Dead` | QueueBox marks the row dead at once. | `422`. QueueBox stores no row. | The same as `Fail`. | A route or destination transform error increases `queuebox_transform_failures_total{strategy}` by one. A source rejection with `Fail` or `Dead` increases `queuebox_inbox_rejections_total{reason="transform_failed"}` by one. ## Source topic template [Section titled “Source topic template”](#source-topic-template) `sources..topic` sets the topic of the outbox row that the relay writes for a push row. | Placeholder | Value | | ----------------- | ------------------------------------------------------ | | `{{ source }}` | The source name. | | `{{ eventType }}` | The `event_type` of the inbox row, or an empty string. | Both placeholders also work without spaces: `{{source}}`. Any other text stays as it is. QueueBox trims the result. The relay marks the inbox row dead when the result is empty. | Source kind | Default | | --------------------------- | ----------------- | | `http` | `{{ eventType }}` | | `rabbitmq`, `kafka`, `nats` | `{{ source }}` | ## Address templates [Section titled “Address templates”](#address-templates) The RabbitMQ `exchange`, the Kafka `topic` and the NATS `subject` accept these placeholders: | Placeholder | Value | | ----------------------- | -------------------------------------------------------------------------------------------------- | | `{{ topic }}` | The `topic` of the outbox row. | | `{{ key }}` | The `key` of the row, or an empty string. | | `{{ aggregateType }}` | The `aggregate_type` of the row, or an empty string. | | `{{ payload. }}` | A field of the payload. A dot separates nested fields. An object or an array renders as JSON text. | | `{{ data. }}` | The same as `payload.`. | Any other placeholder stops the start. The error names the field and the destination. An address that renders empty fails the publish of that row. `exchangeFrom`, `topicFrom` and `subjectFrom` take the value of the `aggregate_type`, `topic` or `key` column, and QueueBox renders no template. ## Routing key templates [Section titled “Routing key templates”](#routing-key-templates) A route `routingKeyTemplate` sets the RabbitMQ routing key or the Kafka record key. A RabbitMQ destination has its own `routingKeyTemplate`, with the default `{{ topic }}`. | Placeholder | Value | | ----------------------- | ------------------------------------------------------ | | `{{ topic }}` | The `topic` of the outbox row. | | `{{ key }}` | The `key` of the row. | | `{{ aggregateType }}` | The `aggregate_type` of the row. | | `{{ payload. }}` | A field of the payload. A dot separates nested fields. | | `{{ data. }}` | The same as `payload.`. | A placeholder works with or without the inner spaces: `{{topic}}` and `{{ topic }}` are the same. A placeholder with another name renders the missing-field default. A field that the row does not hold also renders the default. The default is `routingKeyMissingFieldDefault` of the route, or an empty string. A routing key template does not stop the start on an unknown name. The test [`RoutingKeyTemplateContractTest`](https://github.com/alternayte/queuebox/blob/main/outbox-service/src/test/kotlin/org/nxtspec/RoutingKeyTemplateContractTest.kt) pins these forms: | Template | Payload | Topic | Result | | ---------------------------- | ------------------------------ | ---------------- | ---------------- | | `{{ topic }}` | any | `orders.created` | `orders.created` | | `{{topic}}` | any | `orders.created` | `orders.created` | | `{{ payload.region }}` | `{"region":"eu"}` | any | `eu` | | `{{ data.customer.region }}` | `{"customer":{"region":"de"}}` | any | `de` | | `{{ payload.missingField }}` | `{"region":"eu"}` | any | the default | | `{{ region }}` | `{"region":"eu"}` | any | the default | ### Which template sets the key [Section titled “Which template sets the key”](#which-template-sets-the-key) | Destination | Route sets `routingKeyTemplate` | Route sets none | | ----------- | ------------------------------- | ------------------------------------------------------------------------------------------------------------------ | | RabbitMQ | The route template | The destination `routingKeyTemplate` | | Kafka | The route template | The destination `keyTemplate`. It renders `{{ key }}` and `{{ topic }}` only. An empty result sends no record key. | | HTTP, NATS | not used | not used | # Deliver your first outbox message > Start QueueBox with Docker Compose, insert one outbox row, and watch QueueBox deliver it to an HTTP receiver. In this tutorial you start QueueBox with Docker Compose, insert one row into the outbox table, and watch QueueBox deliver that row to an HTTP receiver. At the end, the row is in state `sent`. ## Before you start [Section titled “Before you start”](#before-you-start) You need Git, Docker with the Compose plugin, and `curl`. The stack uses the ports `8080` and `5432` on your machine, so stop anything else that listens on them. ## Start the stack [Section titled “Start the stack”](#start-the-stack) Clone the repository and start the Compose stack. ```bash git clone https://github.com/alternayte/queuebox.git cd queuebox docker compose -f docker-compose.yml --env-file .env.example up -d --build ``` The stack runs three services: | Service | What it does | | ---------- | -------------------------------------------------------------------------------- | | `postgres` | PostgreSQL 16. QueueBox creates its tables in it at startup. | | `queuebox` | QueueBox, built from the repository. It reads `examples/queuebox.yml`. | | `receiver` | A small HTTP server. It answers `200` and prints every request body it receives. | The first build compiles QueueBox, so it takes some time. Ask QueueBox whether it is ready: ```bash curl http://localhost:8080/health ``` A ready instance answers with status `200` and a body that starts with `{"status":"healthy"`. If `curl` cannot connect, wait a moment and send the request again. Note `-f docker-compose.yml` selects the shipped stack. Without it, Docker Compose also reads `docker-compose.override.yml`, which holds a development loop that this tutorial does not need. Every command below names the file for the same reason. ## Read the route [Section titled “Read the route”](#read-the-route) Open `examples/queuebox.yml`. The configuration names one HTTP destination and a route to it: ```yaml # fragment destinations: webhook-api: type: http baseUrl: http://receiver:8080 path: /webhook timeoutMs: 30000 routes: - topicPattern: "order.*" destination: webhook-api ``` A row whose topic matches `order.*` goes to `webhook-api`. QueueBox sends it as an HTTP `POST` to `http://receiver:8080/webhook`. ## Insert an outbox row [Section titled “Insert an outbox row”](#insert-an-outbox-row) Your application sends a message with one `INSERT` into the `outbox` table. Run that insert through `psql` in the `postgres` container: ```bash docker compose -f docker-compose.yml exec -T postgres psql -U queuebox -d queuebox <<'SQL' BEGIN; INSERT INTO outbox (topic, key, payload) VALUES ('order.created', 'order-1001', '{"orderId":"1001","amount":42}'); COMMIT; SQL ``` `psql` prints `BEGIN`, `INSERT 0 1` and `COMMIT`. The row names three columns: * `topic` is `order.created`, so the route `order.*` matches it. * `key` is `order-1001`. The rows of one key arrive in insert order. * `payload` is the JSON body that the receiver gets. In a real application, the same transaction also writes the business row, for example the order itself. Both rows commit together, or neither commits. ## Watch the delivery [Section titled “Watch the delivery”](#watch-the-delivery) Read the log of the receiver: ```bash docker compose -f docker-compose.yml logs receiver ``` The last line shows the delivery: ```text receiver-1 | receiver listening on 8080 receiver-1 | delivered POST /webhook {"amount":42,"orderId":"1001"} ``` The keys of the body are in a different order. PostgreSQL stores `payload` as `JSONB`, which keeps the content of a JSON object but not the order of its keys. QueueBox polls the outbox every 100 milliseconds in this configuration, so the line appears at once. If the line is absent, run the command again. ## See the row in state `sent` [Section titled “See the row in state sent”](#see-the-row-in-state-sent) Read the row back from the outbox table: ```bash docker compose -f docker-compose.yml exec postgres psql -U queuebox -d queuebox \ -c "SELECT topic, key, state, attempt FROM outbox;" ``` The output shows the row in state `sent`: ```text topic | key | state | attempt ---------------+------------+-------+--------- order.created | order-1001 | sent | 0 (1 row) ``` `attempt` counts the failed deliveries. It is `0`, because the receiver accepted the first delivery. ## Clean up [Section titled “Clean up”](#clean-up) Stop the stack and delete its database volume: ```bash docker compose -f docker-compose.yml down -v ``` ## What you did [Section titled “What you did”](#what-you-did) You started QueueBox, wrote one outbox row, and saw QueueBox deliver it to an HTTP endpoint and mark it `sent`. QueueBox did the polling, the routing and the delivery. The application wrote one row. ## Next steps [Section titled “Next steps”](#next-steps) * [Receive a webhook](/tutorials/receive-a-webhook/) shows the other direction: the inbox. * [Write outbox rows](/how-to/write-outbox-rows/) shows the insert inside a business transaction, with headers, a schedule and a retry ceiling. * [Fan out over HTTP](/how-to/fan-out-over-http/) sends rows to more than one HTTP endpoint. * [Delivery semantics](/concepts/delivery-semantics/) states what QueueBox promises for a row it delivers. # Receive a webhook > Post a webhook to a QueueBox inbox source, see QueueBox reject the duplicate, and consume the stored message with a pull client. In this tutorial you post a webhook to a QueueBox inbox source, post it a second time and see QueueBox reject the duplicate. Then you consume the stored message with a pull client in Go, TypeScript or C#. The client writes an order row and completes the message in one transaction. ## Before you start [Section titled “Before you start”](#before-you-start) You need Git, Docker with the Compose plugin, and `curl`. You also need one of these toolchains: * Go 1.24 or later. * Node 22.18 or later, which runs a TypeScript file directly. * The .NET 8 SDK or later. The stack uses the ports `8080` and `5432` on your machine. If you did [the first tutorial](/tutorials/first-outbox-message/), run `docker compose -f docker-compose.yml down -v` first to start from an empty database. ## Get the repository [Section titled “Get the repository”](#get-the-repository) Clone the repository and change into it. If you cloned it in the first tutorial, change into that directory instead. ```bash git clone https://github.com/alternayte/queuebox.git cd queuebox ``` ## Add a pull source [Section titled “Add a pull source”](#add-a-pull-source) An inbox source is one configured ingress. This tutorial adds an HTTP source named `orders`. Its messages stay in the inbox until your worker takes them, because the source sets `consumption: pull`. `sources` is the last block of `examples/queuebox.yml`. Append the new source to it: ```bash cat >> examples/queuebox.yml <<'YAML' orders: type: http path: /orders idempotencyKeyPath: $.id consumption: pull YAML ``` The source now reads: ```yaml sources: orders: type: http path: /orders idempotencyKeyPath: $.id consumption: pull ``` * `path: /orders` puts the source at `POST /inbox/orders`, because `inbox.basePath` is `/inbox`. * `idempotencyKeyPath: $.id` reads the idempotency key from the `id` field of the body. * `consumption: pull` tells QueueBox not to forward the message. A pull source needs no topic, no route and no destination. Note The Compose file mounts `examples/queuebox.yml` as a single file. Append to the file as the command above does. An editor that writes a new file in its place breaks the mount until you restart the stack. Start the stack: ```bash docker compose -f docker-compose.yml --env-file .env.example up -d --build ``` The first build compiles QueueBox. Ask QueueBox whether it is ready: ```bash curl http://localhost:8080/health ``` A ready instance answers with status `200` and a body that starts with `{"status":"healthy"`. If `curl` cannot connect, wait a moment and send the request again. ## Post the webhook [Section titled “Post the webhook”](#post-the-webhook) Send an order to the source: ```bash curl -i -X POST http://localhost:8080/inbox/orders \ -H 'Content-Type: application/json' \ -d '{"id":"order-1","total":42}' ``` QueueBox stores the message and answers `202 Accepted`. The body holds the identifier of the new inbox row: ```text HTTP/1.1 202 Accepted ... {"messageId":"8d5c3f0e-2f7b-4c0e-9d51-6f0c1b3a7e21"} ``` Your identifier differs. ## Post it again [Section titled “Post it again”](#post-it-again) A webhook sender retries when it does not see an answer, so the same message can arrive twice. Send the same request again: ```bash curl -i -X POST http://localhost:8080/inbox/orders \ -H 'Content-Type: application/json' \ -d '{"id":"order-1","total":42}' ``` QueueBox finds a stored row with the same source and the same idempotency key. It stores nothing and answers `200 OK`: ```text HTTP/1.1 200 OK ... {"status":"duplicate"} ``` ## Look at the stored row [Section titled “Look at the stored row”](#look-at-the-stored-row) Read the inbox table: ```bash docker compose -f docker-compose.yml exec postgres psql -U queuebox -d queuebox \ -c "SELECT source, idempotency_key, consumption, state FROM inbox;" ``` The inbox holds one row, in state `pending`: ```text source | idempotency_key | consumption | state --------+-----------------+-------------+--------- orders | order-1 | pull | pending (1 row) ``` ## Create the business table [Section titled “Create the business table”](#create-the-business-table) The worker writes each order into a table that your application owns. Create it in the same database: ```bash docker compose -f docker-compose.yml exec postgres psql -U queuebox -d queuebox \ -c "CREATE TABLE IF NOT EXISTS orders (id TEXT PRIMARY KEY, total INTEGER NOT NULL);" ``` ## Write the worker [Section titled “Write the worker”](#write-the-worker) The pull client claims the messages of one source. It gives each message to your handler with an open transaction. When the handler returns, the client marks the message `processed` in that same transaction and commits. The order row and the completion commit together, or neither commits. Choose a language, create the project and write the worker. * Go ```bash mkdir orders-worker && cd orders-worker go mod init example.com/orders-worker ``` Write `main.go`: ```go package main import ( "context" "database/sql" "log" "os" "os/signal" "syscall" _ "github.com/jackc/pgx/v5/stdlib" queuebox "github.com/alternayte/queuebox/clients/go" ) func main() { db, err := sql.Open("pgx", "postgres://queuebox:queuebox@localhost:5432/queuebox?sslmode=disable") if err != nil { log.Fatal(err) } defer db.Close() worker, err := queuebox.NewInboxWorker(db, queuebox.Options{Source: "orders"}) if err != nil { log.Fatal(err) } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() log.Println("Waiting for messages of the source 'orders'. Press Ctrl+C to stop.") err = worker.Run(ctx, func(ctx context.Context, message queuebox.Message, tx *sql.Tx) error { var order struct { ID string `json:"id"` Total int `json:"total"` } if err := message.UnmarshalPayload(&order); err != nil { return err } if _, err := tx.ExecContext(ctx, "INSERT INTO orders (id, total) VALUES ($1, $2)", order.ID, order.Total); err != nil { return err } log.Printf("Stored the order %s.", message.IdempotencyKey) return nil }) if err != nil { log.Fatal(err) } } ``` Fetch the dependencies and run it: ```bash go mod tidy go run . ``` * TypeScript ```bash mkdir orders-worker && cd orders-worker npm init -y npm pkg set type=module npm install @alternayte/queuebox-inbox pg ``` Write `worker.ts`: ```ts import pg from "pg"; import { InboxWorker, fromPg } from "@alternayte/queuebox-inbox"; const stopping = new AbortController(); process.on("SIGINT", () => stopping.abort()); process.on("SIGTERM", () => stopping.abort()); const pool = new pg.Pool({ connectionString: "postgres://queuebox:queuebox@localhost:5432/queuebox" }); const worker = new InboxWorker(fromPg(pool), { source: "orders" }); console.log("Waiting for messages of the source 'orders'. Press Ctrl+C to stop."); await worker.run(async (message, tx) => { const order = message.payload as { id: string; total: number }; await tx.query("INSERT INTO orders (id, total) VALUES ($1, $2)", [order.id, order.total]); console.log(`Stored the order ${message.idempotencyKey}.`); }, stopping.signal); await pool.end(); ``` Run it: ```bash node worker.ts ``` * C# ```bash dotnet new console -o OrdersWorker && cd OrdersWorker dotnet add package QueueBox.Inbox dotnet add package Npgsql ``` Replace `Program.cs`: ```csharp using Npgsql; using QueueBox.Inbox; using var stopping = new CancellationTokenSource(); Console.CancelKeyPress += (_, eventArgs) => { eventArgs.Cancel = true; stopping.Cancel(); }; await using var dataSource = NpgsqlDataSource.Create( "Host=localhost;Database=queuebox;Username=queuebox;Password=queuebox"); var worker = new InboxWorker( InboxConnections.From(dataSource), new InboxOptions { Source = "orders" }); Console.WriteLine("Waiting for messages of the source 'orders'. Press Ctrl+C to stop."); await worker.RunAsync(async (message, transaction, cancellationToken) => { await using var command = transaction.CreateCommand(); command.CommandText = "INSERT INTO orders (id, total) VALUES (@id, @total)"; command .WithParameter("@id", message.Payload.GetProperty("id").GetString()) .WithParameter("@total", message.Payload.GetProperty("total").GetInt32()); await command.ExecuteNonQueryAsync(cancellationToken); Console.WriteLine($"Stored the order {message.IdempotencyKey}."); }, stopping.Token); ``` Run it: ```bash dotnet run ``` The worker claims the stored message at once and prints: ```text Stored the order order-1. ``` Leave the worker running. ## Check the result [Section titled “Check the result”](#check-the-result) Open a second terminal in the `queuebox` directory. Read the order table and the inbox row: ```bash docker compose -f docker-compose.yml exec postgres psql -U queuebox -d queuebox \ -c "SELECT id, total FROM orders;" \ -c "SELECT idempotency_key, state FROM inbox WHERE source = 'orders';" ``` The order row exists, and the inbox row is in state `processed`: ```text id | total ---------+------- order-1 | 42 (1 row) idempotency_key | state -----------------+----------- order-1 | processed (1 row) ``` Post a new order while the worker runs, and watch the worker print it: ```bash curl -X POST http://localhost:8080/inbox/orders \ -H 'Content-Type: application/json' \ -d '{"id":"order-2","total":17}' ``` ## Clean up [Section titled “Clean up”](#clean-up) Press Ctrl+C in the worker terminal. The worker stops claiming and lets a running handler finish. Then stop the stack, delete its volume, and restore the configuration file: ```bash docker compose -f docker-compose.yml down -v git checkout examples/queuebox.yml ``` ## What you did [Section titled “What you did”](#what-you-did) You posted a webhook to an inbox source, and QueueBox stored it once although it arrived twice. A pull worker then took the message and wrote an order row. The completion committed in the same transaction as the order row. ## Next steps [Section titled “Next steps”](#next-steps) * [Consume the inbox](/how-to/consume-the-inbox/) compares the push relay with pull workers, and covers retries and idempotent handlers. * [Authenticate requests](/how-to/authenticate-requests/) protects an inbox source with a token, an API key or an HMAC signature. * [Pull clients](/reference/pull-clients/) lists every setting of the three libraries.