---
title: "A transactional pipeline"
description: "Chain stages so that each one acknowledges its input and publishes the next stage's message in a single PostgreSQL transaction, with per-entity ordering preserved end to end."
---

> Documentation Index
> Fetch the complete documentation index at: https://queenmq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# A transactional pipeline

A pipeline stage does two writes: it publishes its output and it consumes its input. If those are separate requests, one of them can be the last thing that happens before a crash, and you get either a lost message or a duplicated one depending on the order you chose. `POST /api/v1/transaction` removes the choice by putting both in one PostgreSQL transaction.

## One stage

Pop from the stage's input queue, do the work, then commit the output push and the input ack together.

That is the whole pattern. The transaction body is a flat `operations` array of `push` and `ack` entries; the client SDKs also collect the lease ids of the messages you ack into a top-level `requiredLeases` list, which is how the broker knows which leases must still be valid for the commit to be allowed.

Only `push` and `ack` operation types exist. Anything else is rejected with HTTP 400 and the message `segments transaction supports only push and ack operations`.

## The chain

Four stages, each one reading the previous stage's output queue:

```js
// Illustrative, not extracted from a test.
await client
  .queue('orders.validated')
  .group('enrich')
  .batch(100)
  .consume(async messages => {
for (const msg of messages) {
  const enriched = await enrich(msg.data)
  await client
    .transaction()
    .queue('orders.enriched')
    .partition(msg.data.orderId)
    .push([{ data: enriched }])
    .ack(msg)
    .commit()
}
  })
```

Two details in that snippet carry the guarantees.

**`partition(msg.data.orderId)` is what preserves order.** Ordering in Queen is per partition, in commit order, and there is no global ordering. Using the entity as the partition name at every stage means every event for one order stays in one lane from end to end, while different orders proceed in parallel. The partition allocator bumps `queen.log_partitions.last_offset` under that row's `FOR UPDATE` lock, which is also what makes `created_at` monotone within a partition, the property timestamp seeks and time-based retention depend on.

**The ack is inside the transaction.** Because an ack is a cursor commit, the stage's cursor and the next stage's message become durable at the same instant. There is no window in which the input is consumed but the output does not exist.

## What all-or-nothing actually means

The transaction procedure runs the whole batch as one PostgreSQL transaction and every failure path raises, which takes the rest of the batch with it. Three failure modes are worth knowing before you build on this.

1. **A duplicate push aborts the entire transaction.** Outside a transaction, a push whose `transactionId` is already inside the deduplication window returns `status: "duplicate"` and writes nothing. Inside a transaction, that same verdict is escalated to a raised error (`QDUP`, classified as a unique violation) and rolls back every other push and ack in the batch. A blind retry of a whole transaction with deterministic transaction ids therefore fails as a unit rather than degrading gracefully.

2. **A rejected ack aborts the entire transaction.** An invalid or expired lease, or a position beyond what was leased, escalates the same way (`QTXN`). If your handler takes longer than the lease, the commit fails and nothing is written: the input redelivers, which is the correct outcome, but the output push is lost with it and must be redone.

3. **A repeat inside one request is first-wins, not an error.** Two push items in the same transaction with the same `transactionId`, queue and partition produce one frame. The second item echoes back with `duplicate: true` and the first item's `messageId`. This is the only place `duplicate` appears in a transaction response.

> **Caution**
>
> A failed transaction answers **HTTP 200** with `{"transactionId": "...", "success": false, "error": "...", "results": []}`. The status code is not the signal; `success` is. The JavaScript, Python and Go SDKs throw on `success: false`, but a raw HTTP client that checks only the status code will read a rolled-back transaction as a successful one.

### It is not exactly-once end to end

The commit is atomic. The response is not. If the connection drops after PostgreSQL commits but before you read the reply, a retry pushes the message again, unless the `transactionId` is deterministic and still inside the deduplication window, in which case the retry aborts as a duplicate (point 1 above) and you know the original landed.

The practical shape for a stage that must not double-publish is: derive the output `transactionId` deterministically from the input message id, and treat a `QDUP` failure on retry as "already done" rather than as an error. See [Idempotent producers](/use/examples/dedup) for how to choose that id and how long the window protects you.

### Concurrent transactions do not deadlock each other

Every multi-partition writer in the broker takes its partition row locks in one total order: ascending `log_partitions.id`, in a single set-based pre-lock. The transaction procedure, the bundled push path and the retention steps all follow it, and the ack path takes only the consumer row lock and never a partition lock, so it cannot form a cycle with the push serializer. You do not need to order your pushes to avoid deadlocks.

## Fan-out from one stage

One transaction can push to several queues and ack several inputs. The verified suite covers pushing to two queues while acking one input, and acking three messages at once; the shape is the same array with more entries:

```js
// Illustrative, not extracted from a test.
await client
  .transaction()
  .queue('orders.enriched')
  .push([{ data: enriched }])
  .queue('orders.audit')
  .push([{ data: { orderId, at: Date.now() } }])
  .ack(msg)
  .commit()
```

Both queues either receive their message or neither does.

## Measured behaviour of this shape

A four-stage pipeline of exactly this construction is the archived ordering result: **1000 partitions, 25,000 events per second sustained for 600 seconds, 88,503,408 messages verified with 0 duplicates, 0 gaps and 0 order violations.** The loader ran with a 300-second deduplication window, which it echoes in its own configuration line, and the artifact is `benchmark-queen/2026-07-23-3test-report/` (test T3). Full conditions and raw output are in [benchmarks](/benchmarks).

## What to expect

| Situation | Result |
| --- | --- |
| Crash between the work and the commit | Nothing committed; input redelivers when the lease expires |
| Crash after the commit, before the response | Output exists and input is acked; a retry aborts as a duplicate if the id is deterministic and in the window |
| Handler slower than the lease | Commit rejected (`QTXN`), whole batch rolled back, input redelivers |
| Same entity, two events, two workers | Both serialize on that partition's allocator lock; order is commit order |
| Different entities | Independent partitions, fully parallel |
| Output queue does not exist yet | Created inside the same transaction as the push |

## Next

- [Deterministic transaction ids](/use/examples/dedup) — Choosing an idempotency key, and what the window protects.
- [Windowed stages](/use/streams) — When a stage needs aggregation over time rather than a message-at-a-time transform.
- [Errors and retries](/use/errors) — Which failures are terminal, which are backpressure, and what the SDKs do about each.

Source: https://queenmq.com/use/examples/pipeline/index.mdx