---
title: "Transactions: acks, pushes and state in one commit"
description: "One call acks the input, pushes to any queue, writes KV and schedules timers, and commits as one log entry, all or nothing. Conditions on KV writes stand in for reads."
---

> Queen MQ documentation, for AI agents
> Complete self-contained summary of Queen MQ: https://queenmq.com/llms-brief.txt
> Fetch that first when the question is about the product rather than about this page.
> Index of all pages: https://queenmq.com/llms.txt

# Transactions: acks, pushes and state in one commit

The ack, the order's new state, the next event and tomorrow's reminder commit as one log entry, or none of them does. That one call replaces the outbox table, the idempotency table and the cron job.

JavaScript:

```js
await queen.queue('orders')
  .group('fulfillment')
  .subscriptionMode('all')
  .autoAck(false)
  .each()
  .limit(1)
  .consume(async (msg) => {
    const { orderId, customer } = msg.data
    await queen.transaction()
      .ack(msg, 'completed', { consumerGroup: 'fulfillment' })  // the input is done
      .kv.put('orders', `order-${orderId}`, { status: 'paid' }, { ttl: '7d' })
      .queue('shipments').partition(msg.partition)              // the next event
      .push([{ transactionId: `ship-${orderId}`, data: { orderId } }])
      .timer('reminders').key(`review-${orderId}`).delay('24h') // a message tomorrow
      .payload({ orderId, customer }).schedule()
      .commit()
  })
```

Python:

```python
from datetime import timedelta

async def fulfil(msg):
    order_id, customer = msg["data"]["orderId"], msg["data"]["customer"]
    await (
        queen.transaction()
        .ack(msg, "completed", {"consumer_group": "fulfillment"})  # the input is done
        .kv.put("orders", f"order-{order_id}", {"status": "paid"}, ttl=timedelta(days=7))
        .queue("shipments").partition(msg["partition"])             # the next event
        .push([{"transactionId": f"ship-{order_id}", "data": {"orderId": order_id}}])
        .timer("reminders").key(f"review-{order_id}").after(timedelta(hours=24))  # tomorrow
        .payload({"orderId": order_id, "customer": customer}).schedule()
        .commit()
    )

await (
    queen.queue("orders")
    .group("fulfillment")
    .subscription_mode("all")
    .auto_ack(False)
    .each()
    .limit(1)
    .consume(fulfil)
)
```

Go:

```go
err = client.Queue("orders").
	Group("fulfillment").
	SubscriptionMode(queen.SubscriptionModeAll).
	AutoAck(false).
	Each().
	Limit(1).
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		orderID := fmt.Sprint(msg.Data["orderId"])
		_, err := client.Transaction().
			Ack(msg, queen.AckStatusCompleted, queen.AckOptions{ConsumerGroup: "fulfillment"}). // the input is done
			KV(queen.KVPutOp("orders", "order-"+orderID,                                        // the order's state
									map[string]interface{}{"status": "paid"}, queen.TTL(7*24*time.Hour))).
			Queue("shipments").Partition(msg.Partition). // the next event
			Push(queen.PushItem{TransactionID: "ship-" + orderID, Payload: map[string]interface{}{"orderId": orderID}}).
			Timers(queen.ScheduleTimerOp(queen.TimerSchedule{ // a message tomorrow
				Queue: "reminders", TimerKey: "review-" + orderID, Delay: 24 * time.Hour,
				Payload: map[string]interface{}{"orderId": orderID, "customer": msg.Data["customer"]},
			})).
			Commit(ctx)
		return err
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
use queen_mq::{Expiry, TxnPushItem};

let client = queen.clone();
queen
    .queue("orders")
    .group("fulfillment")
    .subscription_mode(SubscriptionMode::All)
    .auto_ack(false)
    .limit(1)
    .consume(move |msg: Message| {
        let client = client.clone();
        async move {
            let id = &msg.data["orderId"];
            client
                .transaction()
                .ack(&msg) // the input is done
                .kv_put("orders", &format!("order-{id}"), json!({ "status": "paid" }),
                    Expiry::Seconds(7 * 86400))? // the order's state
                .push_item(TxnPushItem { // the next event, same customer
                    queue: "shipments".into(),
                    partition: Some(msg.partition.clone()),
                    payload: json!({ "orderId": id }),
                    transaction_id: Some(format!("ship-{id}")),
                    trace_id: None,
                })?
                .schedule("reminders", &format!("review-{id}"), Duration::from_secs(86400), // a message tomorrow
                    json!({ "orderId": id, "customer": msg.data["customer"] }).to_string())
                .commit()
                .await?;
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;
```

PHP:

```php
$queen->queue('orders')
    ->group('fulfillment')
    ->subscriptionMode('all')
    ->autoAck(false)
    ->each()
    ->limit(1)
    ->consume(function (array $msg) use ($queen) {
        $id = $msg['data']['orderId'];
        $queen->transaction()
            ->ack($msg, 'completed', ['consumerGroup' => 'fulfillment']) // the input is done
            ->kv('orders')->put("order-{$id}", ['status' => 'paid'], ['ttlSeconds' => 7 * 86400])
            ->queue('shipments')->partition($msg['partition'])           // the next event
            ->push([['transactionId' => "ship-{$id}", 'data' => ['orderId' => $id]]])
            ->timers('reminders')->schedule("review-{$id}", 24 * 3600 * 1000, // tomorrow, in ms
                ['orderId' => $id, 'customer' => $msg['data']['customer']])
            ->commit();
    })
    ->execute();
```

C++:

```cpp
client.queue("orders")
    .group("fulfillment")
    .subscription_mode("all")
    .auto_ack(false)
    .each()
    .limit(1)
    .consume([&](const json& msg) {
        const json& order = msg["data"];
        const std::string id = std::to_string(order["orderId"].get<int>());
        queen::TimerSchedule reminder;                       // a message tomorrow
        reminder.queue = "reminders";
        reminder.timer_key = "review-" + id;
        reminder.delay_millis = 24 * 60 * 60 * 1000;
        reminder.payload = {{"orderId", order["orderId"]},
                            {"customer", order["customer"]}};

        // client.transaction() cannot give its push a partition, so the step goes
        // to the broker's transaction route through the client's own HTTP layer.
        json step = {
            {"operations", json::array({
                {{"type", "ack"}, {"status", "completed"},   // the input is done
                 {"consumerGroup", "fulfillment"}, {"leaseId", msg["leaseId"]},
                 {"transactionId", msg["transactionId"]},
                 {"partitionId", msg["partitionId"]}},
                {{"type", "push"}, {"items", json::array({{  // the next event
                    {"queue", "shipments"}, {"partition", msg["partition"]},
                    {"transactionId", "ship-" + id},
                    {"payload", {{"orderId", order["orderId"]}}},
                }})}},
            })},
            {"requiredLeases", json::array({msg["leaseId"]})},
            {"kv", json::array({queen::wire::kv_write_op("put", "orders", "order-" + id,
                {{"status", "paid"}}, queen::KvTtl::seconds(7 * 86400), {})})},
            {"timers", json::array({queen::wire::timer_schedule_op(reminder)})},
        };
        client.get_http_client()->post("/api/v1/transaction", step);
    });
```

Then open the dashboard: its KV page holds the order's new state, and its Timers page holds tomorrow's reminder.

**Why:** a worker's step is several writes: it marks its input done, saves what changed, tells the next service and plans what comes later. When those writes go to different systems, a crash between two of them leaves the step half done, and someone writes code to find and repair the halves. One commit has no halves.

If the worker took too long and its lease ran out, the whole transaction is refused, so a slow worker can never write over the one that has the message now. If the step committed and only the answer got lost, a retry with the same `transactionId`s is refused whole, and nothing is written twice.

**Figure.** Worker w1 pops a batch and is paused, a long garbage-collection pause, past its lease. The lease runs out and w2 gets the same batch, processes it and commits. When w1 wakes up and commits its transaction (the ack, a KV write and a push), the ack's lease is no longer w1's, so the whole transaction rolls back: the push and the KV write are not written either.

A worker whose lease ran out: its whole transaction rolls back.

## What can go in

| Part | What it can hold |
| --- | --- |
| Acks | Messages leased from any partitions, queues and consumer groups of the tenant. Each ack carries its group and the lease of the pop that delivered it. |
| Pushes | Messages to any queues and partitions. A queue or a partition that does not exist is created, as by a push. |
| KV | `put`, `putIfAbsent`, `delete`, `incr` and `check`, each one optionally with a condition (`expect`, `required`). |
| Timers | `schedule`, which replaces a timer with the same key, and `cancel`. |
| A guard | `.guard(lock)`: the call commits only while the lock is still yours. |

Over HTTP the same call is `POST /api/v1/transaction`: the acks and pushes go in `operations`, the KV and timer parts in the top-level arrays `kv` and `timers`. [Transaction API](/reference/transaction/) has every field.

**Why:** a step can touch anything the tenant owns, and it is still one entry of one log. The tenant lives in one raft group with one leader, so there is no coordinator and no two-phase commit.

## Conditions instead of reads

A transaction is one call, not a session: there is no "read, decide, write". To make a write depend on the state, put a condition on it. `expect` names the version the key must have, and `required: true` makes a missed condition roll back the whole call.

JavaScript:

```js
await queen.queue('order-events')
  .group('workflow')
  .subscriptionMode('all')
  .autoAck(false)
  .each()
  .consume(async (msg) => {
    const { orderId } = msg.data
    const order = await queen.kv.get('orders', orderId)            // { status: 'paid' }
    if (order.value.status !== 'paid') {                   // already moved: just ack
      return queen.ack(msg, 'completed', { group: 'workflow' })
    }

    // Do the work of this step here: charge the card, book the courier, ...

    const result = await queen.transaction()
      .ack(msg, 'completed', { consumerGroup: 'workflow' })
      .kv.put('orders', orderId, { status: 'shipping' },
        { ttl: '30d', expect: order.version, required: true })  // only from what we read
      .queue('shipping').partition(orderId)
      .push([{ transactionId: `ship-${orderId}`, data: { orderId } }])
      .commit()
    if (result.success === false) return  // moved by someone else: nothing written
  })
```

Python:

```python
async def step(msg):
    order_id = msg["data"]["orderId"]
    order = await queen.kv.get("orders", order_id)          # {'status': 'paid'}
    if order["value"]["status"] != "paid":                   # already moved: just ack
        return await queen.ack(msg, "completed", {"group": "workflow"})

    # Do the work of this step here: charge the card, book the courier, ...

    result = await (
        queen.transaction()
        .ack(msg, "completed", {"consumer_group": "workflow"})
        .kv.put("orders", order_id, {"status": "shipping"},
                ttl=timedelta(days=30), expect=order["version"],
                required=True)                               # only from what we read
        .queue("shipping").partition(order_id)
        .push([{"transactionId": f"ship-{order_id}", "data": {"orderId": order_id}}])
        .commit()
    )
    if not result:                    # moved by someone else: nothing written
        return

await (
    queen.queue("order-events")
    .group("workflow")
    .subscription_mode("all")
    .auto_ack(False)
    .each()
    .consume(step)
)
```

Go:

```go
err = client.Queue("order-events").
	Group("workflow").
	SubscriptionMode(queen.SubscriptionModeAll).
	AutoAck(false).
	Each().
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		opts := queen.AckOptions{ConsumerGroup: "workflow"}
		orderID := msg.Data["orderId"].(string)
		order, err := client.KV().Get(ctx, "orders", orderID) // Value {"status":"paid"}
		if err != nil {
			return err
		}
		var state struct{ Status string }
		if err := json.Unmarshal(order.Value, &state); err != nil {
			return err
		}
		if state.Status != "paid" { // already moved: just ack
			_, err := client.Ack(ctx, msg, true, opts)
			return err
		}

		// Do the work of this step here: charge the card, book the courier, ...

		next := map[string]interface{}{"status": "shipping"}
		result, err := client.Transaction().
			Ack(msg, queen.AckStatusCompleted, opts).
			KV(queen.KVPutOp("orders", orderID, next, queen.TTL(30*24*time.Hour),
				queen.KVWriteOptions{
					Expect:   queen.Expect(order.Version), // only from what we read
					Required: true,
				})).
			Queue("shipping").Partition(orderID).
			Push(queen.PushItem{TransactionID: "ship-" + orderID,
				Payload: map[string]interface{}{"orderId": orderID}}).
			Commit(ctx)
		if err != nil {
			return err
		}
		if result.IsKVPrecondition() {
			fmt.Println(orderID, "moved by someone else: nothing written")
		}
		return nil
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
let client = queen.clone();
queen
    .queue("order-events")
    .group("workflow")
    .subscription_mode(SubscriptionMode::All)
    .auto_ack(false)
    .consume(move |msg: Message| {
        let queen = client.clone();
        async move {
            let order_id = msg.data["orderId"].as_str().unwrap_or_default().to_string();
            let order = queen.kv().get("orders", &order_id).await?; // {"status":"paid"}
            if order.value.as_ref().and_then(|v| v["status"].as_str()) != Some("paid") {
                queen.ack(&msg).await?; // already moved: just ack
                return Ok(());
            }

            // Do the work of this step here: charge the card, book the courier, ...

            let thirty_days = Expiry::seconds(30 * 86400);
            let shipping = queen
                .kv()
                .put("orders", &order_id, json!({ "status": "shipping" }), thirty_days)
                .expect(order.version.unwrap_or(0)) // only from what we read
                .required()
                .operation()?;
            let result = queen
                .transaction()
                .ack(&msg)
                .kv(shipping)
                .push_item(TxnPushItem {
                    queue: "shipping".into(),
                    partition: Some(order_id.clone()),
                    payload: json!({ "orderId": order_id }),
                    transaction_id: Some(format!("ship-{order_id}")),
                    trace_id: None,
                })?
                .commit()
                .await?;
            if !result.success {
                return Ok(()); // moved by someone else: nothing written
            }
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;
```

PHP:

```php
$queen->queue('order-events')
    ->group('workflow')
    ->subscriptionMode('all')
    ->autoAck(false)
    ->each()
    ->consume(function (array $msg) use ($queen) {
        $orderId = $msg['data']['orderId'];
        $order = $queen->kv()->get('orders', $orderId);       // value: [status => paid]
        if ($order['value']['status'] !== 'paid') {           // already moved: just ack
            $queen->ack($msg, 'completed', ['group' => 'workflow']);
            return;
        }

        // Do the work of this step here: charge the card, book the courier, ...

        $result = $queen->transaction()
            ->ack($msg, 'completed', ['consumerGroup' => 'workflow'])
            ->kv('orders')->put($orderId, ['status' => 'shipping'], [
                'ttlSeconds' => 30 * 86400,
                'expect' => $order['version'],                // only from what we read
                'required' => true,
            ])
            ->queue('shipping')->partition($orderId)->push([
                ['transactionId' => "ship-{$orderId}", 'data' => ['orderId' => $orderId]],
            ])
            ->commit();
        if ($result['success'] === false) {
            return;                        // moved by someone else: nothing written
        }
    })
    ->execute();
```

C++:

```cpp
client.queue("order-events")
    .group("workflow")
    .subscription_mode("all")
    .auto_ack(false)
    .each()
    .consume([&](const json& msg) {
        const std::string order_id = msg["data"]["orderId"];
        json order = client.kv("orders").get(order_id);  // value: {"status":"paid"}
        if (order["value"]["status"] != "paid") {        // already moved: just ack
            client.ack(msg, true, {{"group", "workflow"}});
            return;
        }

        // Do the work of this step here: charge the card, book the courier, ...

        queen::KvWriteOptions from_what_we_read;         // only from what we read
        from_what_we_read.expect = order["version"].get<long long>();
        from_what_we_read.required = true;
        // client.transaction() cannot give its push a partition, so the step goes
        // to the broker's transaction route through the client's own HTTP layer.
        json step = {
            {"operations", json::array({
                {{"type", "ack"}, {"status", "completed"},
                 {"consumerGroup", "workflow"}, {"leaseId", msg["leaseId"]},
                 {"transactionId", msg["transactionId"]},
                 {"partitionId", msg["partitionId"]}},
                {{"type", "push"}, {"items", json::array({{
                    {"queue", "shipping"}, {"partition", order_id},
                    {"transactionId", "ship-" + order_id},
                    {"payload", {{"orderId", order_id}}},
                }})}},
            })},
            {"requiredLeases", json::array({msg["leaseId"]})},
            {"kv", json::array({queen::wire::kv_write_op("put", "orders", order_id,
                {{"status", "shipping"}}, queen::KvTtl::seconds(30 * 86400),
                from_what_we_read)})},
        };
        json result = client.get_http_client()->post("/api/v1/transaction", step);
        bool committed = result.value("success", false);
        if (!committed) return;  // moved by someone else: nothing written
    });
```

`once(ns, key, { ttl })` (JavaScript and Python) is the short form of the most common condition: a marker written with `putIfAbsent` and `required: true`. Key it on the business fact (`paymentId`), and a second delivery of the same fact rolls back whole, receipts and timers included. The other SDKs write the same marker with KV `putIfAbsent` and `required: true`.

**Why:** two workers can read the same version and both decide to write. A condition checked inside the commit cannot race: one of them wins, and the other rolls back with nothing written.

## Why a transaction rolls back

A rollback answers HTTP 200 with `success: false`, a `reason`, and nothing written.

| `reason` | Cause |
| --- | --- |
| `rejected_ack` | An ack's lease ran out or moved to another worker. The message goes to whoever holds it now. |
| `kv_precondition` | A condition with `required: true` failed: a `once` marker that exists, an `expect` that missed, a lock that is gone. |
| `duplicate` | A pushed `transactionId` is already in its partition's dedup window: the step already committed. |
| `reserved` | Another write holds that cursor at this moment. Try again. |
| `bad_request`, `too_large` | A malformed operation, or a part over the entry limit. |

The JavaScript client returns a lost condition from `commit()` and throws for the other reasons. A 503 is not a rollback: `retry`, `timeout` and `no_leader` mean the outcome is not known, so the call may still commit. Make every step safe to send twice, with a `once` marker or a push with a fixed `transactionId`.

**Why:** a lost answer and a lost race look the same to a worker. The reason tells them apart: some mean "already done", some "not yours any more", some "try again".

## Limits

- One transaction belongs to one tenant.
- KV inside a transaction: at most 64 operations over 256 keys, one write per key.
- One timer operation per queue and timer key in one call.
- A request body is at most 64 MiB (`QUEEN_MAX_BODY_BYTES`).
- An external call (a payment provider, an email) is outside the commit. Give it an idempotency key from the message.

## See also

- [Workflows](/patterns/workflows/): an order moving from step to step, one transaction each.
- [KV, locks and timers](/writing-applications/kv-locks-timers/): the parts a transaction writes.
- [Errors](/reference/errors/): every reason and status code.

Source: https://queenmq.com/writing-applications/transactions/index.mdx