---
title: "Workflow pattern: one step, one transaction"
description: "An order moves from paid to shipped: each step acks its input, writes the order's state with a version check and pushes the next event, in one transaction."
---

> 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

# Workflow pattern: one step, one transaction

An order moves through states: paid, shipping, shipped. A redelivered event, two workers, or an event that arrives late must never move it twice or backwards, and a step must never be half done.

The order's state lives in KV. Each step is one transaction: it acks the event, writes the new state only if the state is still the version the worker read (`expect`), and pushes the next event. If the version moved, the whole transaction rolls back, and nothing is written.

**Figure.** The state of order-2001 lives in KV: paid, at version 1. The workflow step reads it, does its work, and commits one transaction: the ack of the event, the KV write to shipping only if the version is still 1, and the push of the next event to shipping. The same step sent again from the old version finds version 2: the whole transaction rolls back and nothing is written, so there is still one shipping event.

A state machine whose steps cannot race.

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
    });
```

The step moved `order-2001` to `shipping` and pushed one shipping event. The same step sent again from the old version rolled back: still one event.

**Why:** the ack, the new state and the next event commit together or not at all. That replaces the outbox table, the idempotency table and the lock you would put around them.

One thing to know: a rollback is an answer, not an error. `commit()` returns `success: false` (in Go, `IsKVPrecondition()` on the result) and does not throw. Read it.

Built from: [Transactions](/writing-applications/transactions/), [KV, locks and timers](/writing-applications/kv-locks-timers/).

Source: https://queenmq.com/patterns/workflows/index.mdx