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.

await client
    .transaction()
    .queue('test-queue-v2-txn-basic-b')
    .push([{ data: { value: messages[0].data.value + 1 } }])
    .ack(messages[0])
    .commit()

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:

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

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

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

What to expect

Next