---
title: "Saga with a compensating timer"
description: "A booking whose hold, compensation timer, payment request and acknowledgement commit as one thing, and a compensating consumer that checks the saga's state before it unwinds anything."
---

> 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

# Saga with a compensating timer

A hotel was sold out on paper for nine nights it had spent empty. The booking system held inventory
when a reservation started and released it when the payment either settled or failed, and the
release was a sleeping task inside the worker process. A rolling deploy replaced the workers. Every
hold in flight lost its release, and nobody noticed for a fortnight, because a hold that is never
released looks exactly like a room that was sold.

The release was not slow. It was in the wrong place. A compensation is not a timeout, it is an
obligation, and an obligation has to outlive the process that took it on. The correction is to make
it a row in the same database as the queue, written in the same transaction as the work it
compensates for. Then the deploy is irrelevant: if the room is held, the compensation exists, and if
the room is not held, nothing else happened either.

The program builds that as five things in one commit: a gate that refuses a booking already held,
the saga's first state, the compensation timer, the payment request, and the acknowledgement of the
submission that started it. It measures four properties. A duplicate submission produces exactly one
payment, one timer and one state row. A compensation cancelled inside a confirming bundle is never
delivered. The one booking whose card was declined is unwound by its timer, with no process awake to
do it. And the booking whose cancel was deliberately skipped is compensated by nobody, because the
consumer refused.

That last one is the mistake this page exists to prevent, and it runs the opposite way to the one
people expect. The cancel is not the safety mechanism. A fired timer leaves no tombstone, so a cancel
that arrives a millisecond after the fire answers `absent` with `ok: false`, which means "no longer
pending" and may mean "already delivered". A design that trusts the cancel unwinds a reservation that
has already shipped, and reports success while doing it. The compensating consumer has to read the
saga's state and decide for itself.

The other application of these two primitives together, a counter that admits and a timer that moves
the refused work to the next window, is already written up on
[the rate limiter page](/use/full-examples/rate-limiter): this one takes the same pair down the
longer axis, where the state is a workflow with steps and the timer is the deadline on it.

### JavaScript

```js title="examples/apps/js/saga.mjs"
//
// A booking saga whose compensation is a timer, and whose every step is a row
// in the same PostgreSQL as the queue.
//
// The war story is a room hold that never came back. A booking system held
// inventory when a reservation started and released it when the payment either
// settled or failed, and the release lived in a setTimeout inside the worker. A
// rolling deploy replaced the workers; every hold in flight lost its release;
// and a fortnight later somebody noticed a hotel had been sold out on paper for
// nine nights it had spent empty.
//
// The release was not slow, it was in the wrong place. A compensation is not a
// timeout, it is an obligation, and an obligation has to outlive the process
// that took it on. Here the gate, the saga state, the compensation timer, the
// payment request and the acknowledgement are ONE PostgreSQL transaction. If
// the room is held, the compensation exists. If the room is not held, nothing
// else happened either. There is no interval in between for a deploy to land
// in.
//
//   bookings
//     └── group "reserver"   ONE bundle: gate + state + timer + push + ack
//           ├── payments (partitioned by booking)
//           │     └── group "payer"    confirm + CANCEL the timer + ack, one bundle
//           └── expiries (delivered by the timer, at the hold's expiry)
//                 └── group "compensator"   reads the state BEFORE compensating
//
// Run it:
//   QUEEN_URL=http://localhost:6632 node saga.mjs

import { Queen } from 'queen-mq'

const QUEEN_URL = process.env.QUEEN_URL || 'http://localhost:6632'
const RUN = Date.now().toString(36)

// A suffix on the queue names AND on the KV namespace. The queues need it
// because delete-then-recreate leaves stale partition state for up to 30
// seconds; the namespace needs it for the opposite reason -- a saga row
// outlives the run that wrote it, so a second run under the same namespace
// would find every booking already held and measure nothing.
const BOOKINGS = `app-js-saga-bookings-${RUN}`
const PAYMENTS = `app-js-saga-payments-${RUN}`
const EXPIRIES = `app-js-saga-expiries-${RUN}`
const NS = `app-js-saga-${RUN}`

// How long a room stays held before the compensation fires. Short enough for a
// test, and in production it is the only number that changes.
//
// It has to outlast the reserve and pay phases, or a timer would fire before
// the payment that cancels it and the run would be measuring a race rather than
// a design. deliverAt is a floor and never a ceiling, so a timer can only be
// late: a margin here is sound, where a margin in the other direction would not
// be.
const HOLD_MS = 15_000

// Every phase below ends on a COUNT, and this is the deadline behind the count.
// Never wait for silence: a phase that stops when nothing has arrived for a
// while passes on a broker that delivered nothing at all. Reaching the deadline
// ends the phase short, and the count check that follows is what reports it.
const PHASE_MS = 20_000

// The compensation phase gets its own, longer deadline. A timer fires no
// earlier than its delay plus one sweeper cycle, and a broker whose timer table
// has been empty for a while wakes up lazily.
const TIMER_DEADLINE_MS = 90_000

// Four bookings, five submissions. B-2 is submitted twice: the same booking,
// two messages, which is what a redelivery looks like from the reserver's side
// and the reason the bundle opens with a gate rather than with a check.
const BOOKINGS_IN = [
  { bookingId: 'B-1', room: '101', cents: 24000 },
  { bookingId: 'B-2', room: '102', cents: 31000 },
  { bookingId: 'B-2', room: '102', cents: 31000 },
  { bookingId: 'B-3', room: '103', cents: 18000 },
  { bookingId: 'B-4', room: '104', cents: 27000 },
]
const BOOKING_IDS = ['B-1', 'B-2', 'B-3', 'B-4']

// B-3's card is declined, so its saga never reaches "confirmed" and the timer
// is the thing that gives the room back. It is the compensation actually doing
// its job.
const DECLINED = 'B-3'

// B-4 pays, but its cancel is deliberately skipped, which makes the race
// deterministic: a cancel that arrives after the fire answers `absent`, and
// ABSENT MAY MEAN ALREADY DELIVERED. So the compensation for a confirmed
// booking has to be refused by the consumer that receives it, never prevented
// by the cancel alone.
const CANCEL_SKIPPED = 'B-4'

let checks = 0
const assert = (condition, description) => {
  if (!condition) throw new Error(description)
  checks++
  console.log(`  ok: ${description}`)
}

// The saga's state key. It derives from the booking id, which is also the
// partition key of the payments queue: that is what makes the payer's
// read-then-write safe, and it is stated here because it is a property of the
// naming and nothing enforces it.
const sagaKey = (bookingId) => `saga:${bookingId}`

const reserveDecisions = []
const paymentsRequested = []
const compensationsDelivered = []
const roomsReleased = []
const compensationsRefused = []
let preconditionsLost = 0

const queen = new Queen({ url: QUEEN_URL, handleSignals: false })

try {
  console.log(`broker ${QUEEN_URL}`)

  for (const q of [BOOKINGS, PAYMENTS, EXPIRIES]) {
    await queen.queue(q).config({ leaseTime: 30, retryLimit: 3 }).create()
  }

  // ----------------------------------------------------------------- queuing
  console.log('\nsubmitting bookings')
  for (const [i, booking] of BOOKINGS_IN.entries()) {
    await queen.queue(BOOKINGS).push({
      // Distinct transaction ids on purpose. Deduplication would swallow the
      // duplicate submission and the gate would never be tested, and a real
      // redelivery arrives with an identity of its own too.
      transactionId: `submit-${i}-${booking.bookingId}`,
      data: booking,
    })
  }
  console.log(`  ${BOOKINGS_IN.length} submissions for ${BOOKING_IDS.length} bookings`)

  // --------------------------------------------------------------- reserving
  //
  // The bundle, and the whole point of the example: five things commit
  // together, so there is no ordering between them left to get wrong.
  console.log('\nreserving')
  await queen
    .queue(BOOKINGS)
    .group('reserver')
    .subscriptionMode('all')
    .autoAck(false)
    .each()
    // The count that ends the phase, with the deadline behind it.
    .limit(BOOKINGS_IN.length)
    .idleMillis(PHASE_MS)
    .consume(async (msg) => {
      const { bookingId, room, cents } = msg.data

      const res = await queen
        .transaction()
        // 1. The gate AND the first state, in one row. `required: true` is what
        //    makes it a gate instead of a verdict: without it a lost race would
        //    come back applied:false while the payment and the timer went out
        //    anyway.
        .kv.putIfAbsent(NS, sagaKey(bookingId), { step: 'held', room, cents }, { ttl: '1h', required: true })
        // 2. The obligation. From the moment this commits it is a row in the
        //    broker's own table, so it survives this handler, this process,
        //    this deploy and this machine. The key is chosen by us, which is
        //    the entire reason it can be cancelled later by name.
        .timer(EXPIRIES).key(bookingId).delayMs(HOLD_MS).payload({ bookingId, room }).schedule()
        // 3. The work. Partitioned by booking, so every message about one
        //    booking is in one lane.
        .queue(PAYMENTS).partition(bookingId).push({ transactionId: `pay-${bookingId}`, data: { bookingId, cents } })
        // 4. The acknowledgement, carrying this delivery's lease. An expired
        //    lease refuses the ack and takes the other three down with it,
        //    which is the guarantee that no compare-and-swap can give.
        .ack(msg, 'completed', { consumerGroup: msg.consumerGroup })
        .commit()

      reserveDecisions.push(bookingId)

      // A lost gate is RETURNED, not thrown: HTTP 200, success:false, reason
      // "kv_precondition". It is the ordinary outcome of every legitimate
      // redelivery, which makes it one of the most frequent answers this
      // product gives, and it does not belong in a catch block where the
      // reflex is to retry.
      if (res.success === false && res.reason === 'kv_precondition') {
        // Nothing was written: no second payment, no second timer, no second
        // row. The message still has to leave the cursor, so it is
        // acknowledged on its own.
        preconditionsLost++
        await queen.ack(msg, 'completed', { group: msg.consumerGroup })
        console.log(`  ${bookingId}: already held, whole bundle rolled back (${res.kvReason})`)
        return
      }

      console.log(`  ${bookingId}: room ${room} held, compensation armed for ${HOLD_MS} ms`)
    })

  assert(
    reserveDecisions.length === BOOKINGS_IN.length,
    `the reserver reached a decision on every submission (${BOOKINGS_IN.length}, got ${reserveDecisions.length})`
  )
  assert(preconditionsLost === 1, 'the duplicate submission lost the gate exactly once')

  // Pending timers are a table you can read, not a promise you have to trust.
  const armed = await queen.timer(EXPIRIES).list({ limit: 50 })
  console.log(`  timers armed: ${armed.rows.map(r => r.timerKey).sort().join(', ')}`)
  assert(
    armed.rows.length === BOOKING_IDS.length,
    `one compensation is armed per booking and the duplicate added none (${BOOKING_IDS.length}, got ${armed.rows.length})`
  )

  // ------------------------------------------------------------------ paying
  //
  // The other end of the saga. A settled payment confirms the state and calls
  // the compensation off in one commit; a declined card leaves the state where
  // it is and lets the timer do its work.
  console.log('\npaying')
  await queen
    .queue(PAYMENTS)
    .group('payer')
    .subscriptionMode('all')
    .autoAck(false)
    .each()
    .limit(BOOKING_IDS.length)
    .idleMillis(PHASE_MS)
    .consume(async (msg) => {
      const { bookingId } = msg.data
      paymentsRequested.push(bookingId)

      // A read in one call and a write in the next. It is safe HERE because
      // the key derives from the partition key: every message about this
      // booking arrives in one lane of this queue, and a lane has one reader
      // per group. Where a key does not derive from the partition key, this
      // shape is a race and the atomics are the answer -- which is exactly the
      // compensator's situation further down.
      const state = await queen.kv.get(NS, sagaKey(bookingId))

      if (bookingId === DECLINED) {
        // A declined card is a business outcome, not a delivery failure: the
        // message is done with. The room stays held, and nothing in this
        // process is responsible for giving it back.
        await queen.ack(msg, 'completed', { group: msg.consumerGroup })
        console.log(`  ${bookingId}: card declined, hold left to expire`)
        return
      }

      const tx = queen
        .transaction()
        // `expect` makes the serialisation assumption falsifiable instead of
        // silent. If the lane really serialises, it never fails and costs
        // nothing; the day it fails, two consumers are serving one partition
        // and you learn it as a verdict rather than as a wrong total.
        .kv.put(NS, sagaKey(bookingId), { ...state.value, step: 'confirmed' }, {
          ttl: '1h',
          expect: state.version,
          required: true,
        })

      if (bookingId !== CANCEL_SKIPPED) {
        // The cancel rides the bundle. Either the booking is confirmed and the
        // compensation is called off, or neither of the two happened.
        tx.timer(EXPIRIES).key(bookingId).cancel()
      }

      const res = await tx.ack(msg, 'completed', { consumerGroup: msg.consumerGroup }).commit()
      if (res.success === false) throw new Error(`${bookingId}: confirmation lost its fence (${res.kvReason})`)

      console.log(
        `  ${bookingId}: paid and confirmed` +
        (bookingId === CANCEL_SKIPPED ? ', compensation deliberately NOT cancelled' : ', compensation cancelled')
      )
    })

  assert(
    paymentsRequested.length === BOOKING_IDS.length,
    `every booking was asked to pay once and the duplicate produced no second payment ` +
    `(${BOOKING_IDS.length}, got ${paymentsRequested.length})`
  )
  assert(new Set(paymentsRequested).size === paymentsRequested.length, 'no booking was asked to pay twice')

  // The cancel is observable before anything is delivered: the row is gone from
  // the staging table. A peek is how you ask, and a miss is {found:false} with
  // HTTP 200, never a 404.
  const peeked = Object.fromEntries(
    await Promise.all(BOOKING_IDS.map(async id => [id, (await queen.timer(EXPIRIES).key(id).peek()).found]))
  )
  assert(peeked['B-1'] === false, 'the compensation cancelled inside the confirming bundle is gone from the table')
  assert(peeked[DECLINED] === true, `${DECLINED} was never confirmed, so its compensation is still armed`)
  assert(peeked[CANCEL_SKIPPED] === true, `${CANCEL_SKIPPED} is confirmed but its compensation is still armed on purpose`)

  // ------------------------------------------------------------ compensating
  //
  // What the timers deliver, and the consumer that must not trust them.
  //
  // A compensation message is not an instruction, it is a question: is this
  // saga still open? A fired timer leaves no tombstone, so a cancel that
  // arrives a millisecond late answers `absent` and the message goes out
  // anyway. The state is the authority and it is read first.
  //
  // And here the key does NOT derive from the partition key: this message
  // arrives on another queue entirely, in a lane that has nothing to do with
  // the payments lane, so no partitioning could serialise the two writers. That
  // is what `expect` is for, and on this path it is load-bearing rather than an
  // assertion.
  console.log('\ncompensating')
  const compensate = (limit, idleMillis) => queen
    .queue(EXPIRIES)
    .group('compensator')
    .subscriptionMode('all')
    .autoAck(false)
    .each()
    .limit(limit)
    .idleMillis(idleMillis)
    .consume(async (msg) => {
      const { bookingId, room } = msg.data
      compensationsDelivered.push(bookingId)

      const state = await queen.kv.get(NS, sagaKey(bookingId))

      if (!state.found || state.value.step !== 'held') {
        // The booking was confirmed before this fired. Compensating here is
        // how a saga unwinds a sale that has already shipped.
        compensationsRefused.push(bookingId)
        await queen.ack(msg, 'completed', { group: msg.consumerGroup })
        console.log(`  ${bookingId}: state is ${state.value?.step ?? 'gone'}, compensation refused`)
        return
      }

      const res = await queen
        .transaction()
        .kv.put(NS, sagaKey(bookingId), { ...state.value, step: 'expired' }, {
          ttl: '1h',
          expect: state.version,
          required: true,
        })
        .ack(msg, 'completed', { consumerGroup: msg.consumerGroup })
        .commit()

      if (res.success === false) {
        // Somebody confirmed it between the read and the commit. The fence
        // held, nothing was written, and the room stays sold.
        compensationsRefused.push(bookingId)
        await queen.ack(msg, 'completed', { group: msg.consumerGroup })
        console.log(`  ${bookingId}: confirmed under us, compensation refused by the fence`)
        return
      }

      roomsReleased.push(room)
      console.log(`  ${bookingId}: hold expired, room ${room} released`)
    })

  // Two timers were left armed, so two messages must arrive: that is the count,
  // and TIMER_DEADLINE_MS is the deadline behind it.
  await compensate(2, TIMER_DEADLINE_MS)
  assert(
    compensationsDelivered.length === 2,
    `both uncancelled compensations were delivered (2, got ${compensationsDelivered.length}` +
    `${compensationsDelivered.length ? `: ${compensationsDelivered.join(', ')}` : ''})`
  )

  // Then a bounded second pass with room for two more. It is the only honest
  // way to say "a cancelled timer never arrived": the first pass would have
  // stopped at two whatever those two were, so the claim is really that nothing
  // else shows up afterwards.
  await compensate(2, 4000)

  // ---------------------------------------------------------------- checking
  console.log('\nchecking')

  assert(
    compensationsDelivered.length === 2,
    `nothing else arrived on a second pass: still 2 compensations (got ${compensationsDelivered.length})`
  )
  assert(
    !compensationsDelivered.includes('B-1') && !compensationsDelivered.includes('B-2'),
    'a cancelled compensation was never delivered'
  )
  assert(
    roomsReleased.length === 1 && roomsReleased[0] === '103',
    `exactly one room went back on sale, the one whose card was declined (got ${roomsReleased.join(', ') || 'none'})`
  )
  assert(
    compensationsRefused.length === 1 && compensationsRefused[0] === CANCEL_SKIPPED,
    'the compensation for the confirmed booking was refused by the consumer, not prevented by the cancel'
  )

  const states = await queen.kv.getMany(NS, BOOKING_IDS.map(sagaKey))
  assert(
    states.rows.length === BOOKING_IDS.length && states.missing.length === 0,
    `every booking left exactly one saga row (${BOOKING_IDS.length}, got ${states.rows.length})`
  )

  const step = Object.fromEntries(states.rows.map(r => [r.key.replace('saga:', ''), r.value.step]))
  assert(step['B-1'] === 'confirmed' && step['B-2'] === 'confirmed', 'the two ordinary bookings ended confirmed')
  assert(step[CANCEL_SKIPPED] === 'confirmed', `${CANCEL_SKIPPED} is still confirmed after its compensation was delivered`)
  assert(step[DECLINED] === 'expired', `${DECLINED} was unwound by its timer, with nobody awake to do it`)

  console.log(`\n  final: ${Object.entries(step).map(([k, v]) => `${k}=${v}`).sort().join(', ')}`)

  console.log(`\nPASS: ${checks} checks`)
} catch (err) {
  console.error(`\nFAIL: ${err.message}`)
  process.exitCode = 1
} finally {
  // ------------------------------------------------------------------- purge
  //
  // Three things to remove, and the first two are the ones that are easy to
  // forget. The saga rows live in their own table, and a pending timer lives in
  // the staging table keyed by NAME: neither is reached by deleting the queue,
  // and a timer whose queue no longer exists still fires and provisions it
  // again on the way out.
  //
  // Unconditional, in a finally, because a run that FAILED is exactly the run
  // whose leftovers matter: an armed timer would deliver into the next run and
  // a surviving saga row would make the next run pass without holding anything.
  //
  // Best effort throughout: a purge that threw would replace the real verdict
  // with its own.
  try {
    for (const bookingId of BOOKING_IDS) {
      await queen.timer(EXPIRIES).key(bookingId).cancel()
      await queen.kv.delete(NS, sagaKey(bookingId))
    }
    for (const q of [BOOKINGS, PAYMENTS, EXPIRIES]) await queen.queue(q).delete()
  } catch (err) {
    console.error(`  (purge incomplete: ${err.message})`)
  }
  await queen.close()
}
```
### Python

```python title="examples/apps/py/saga.py"
#
# A booking saga whose compensation is a timer, and whose every step is a row in
# the same PostgreSQL as the queue.
#
# The war story is a room hold that never came back. A booking system held
# inventory when a reservation started and released it when the payment either
# settled or failed, and the release lived in a sleeping task inside the worker.
# A rolling deploy replaced the workers; every hold in flight lost its release;
# and a fortnight later somebody noticed a hotel had been sold out on paper for
# nine nights it had spent empty.
#
# The release was not slow, it was in the wrong place. A compensation is not a
# timeout, it is an obligation, and an obligation has to outlive the process
# that took it on. Here the gate, the saga state, the compensation timer, the
# payment request and the acknowledgement are ONE PostgreSQL transaction. If the
# room is held, the compensation exists. If the room is not held, nothing else
# happened either.
#
#   bookings
#     `-- group "reserver"    ONE bundle: gate + state + timer + push + ack
#           |-- payments (partitioned by booking)
#           |     `-- group "payer"    confirm + CANCEL the timer + ack, one bundle
#           `-- expiries (delivered by the timer, at the hold's expiry)
#                 `-- group "compensator"   reads the state BEFORE compensating
#
# Run it:
#   QUEEN_URL=http://localhost:6632 python3 saga.py

import asyncio
import os
import sys
import time
from datetime import timedelta

from queen import Queen

QUEEN_URL = os.environ.get("QUEEN_URL", "http://localhost:6632")

# A suffix on the queue names AND on the KV namespace. The queues need it
# because delete-then-recreate leaves stale partition state for up to 30
# seconds; the namespace needs it for the opposite reason -- a saga row outlives
# the run that wrote it, so a second run under the same namespace would find
# every booking already held and measure nothing.
RUN = f"{int(time.time() * 1000):x}"
BOOKINGS = f"app-py-saga-bookings-{RUN}"
PAYMENTS = f"app-py-saga-payments-{RUN}"
EXPIRIES = f"app-py-saga-expiries-{RUN}"
NS = f"app-py-saga-{RUN}"

# How long a room stays held before the compensation fires. It has to outlast
# the reserve and pay phases, or a timer would fire before the payment that
# cancels it and the run would be measuring a race rather than a design.
# deliverAt is a floor and never a ceiling, so a timer can only be late: a
# margin here is sound, where a margin the other way would not be.
HOLD_MS = 15000

# Every phase below ends on a COUNT, and this is the deadline behind the count.
# Never wait for silence: a phase that stops when nothing has arrived for a
# while passes on a broker that delivered nothing at all.
PHASE_MS = 20000

# The compensation phase gets its own, longer deadline. A timer fires no earlier
# than its delay plus one sweeper cycle, and a broker whose timer table has been
# empty for a while wakes up lazily.
TIMER_DEADLINE_MS = 90000

# Four bookings, five submissions. B-2 is submitted twice: the same booking, two
# messages, which is what a redelivery looks like from the reserver's side and
# the reason the bundle opens with a gate rather than with a check.
BOOKINGS_IN = [
    {"bookingId": "B-1", "room": "101", "cents": 24000},
    {"bookingId": "B-2", "room": "102", "cents": 31000},
    {"bookingId": "B-2", "room": "102", "cents": 31000},
    {"bookingId": "B-3", "room": "103", "cents": 18000},
    {"bookingId": "B-4", "room": "104", "cents": 27000},
]
BOOKING_IDS = ["B-1", "B-2", "B-3", "B-4"]

# B-3's card is declined, so its saga never reaches "confirmed" and the timer is
# the thing that gives the room back.
DECLINED = "B-3"

# B-4 pays, but its cancel is deliberately skipped, which makes the race
# deterministic: a cancel that arrives after the fire answers `absent`, and
# ABSENT MAY MEAN ALREADY DELIVERED. So the compensation for a confirmed booking
# has to be refused by the consumer that receives it, never prevented by the
# cancel alone.
CANCEL_SKIPPED = "B-4"

CHECKS = 0


def check(condition: bool, description: str) -> None:
    """Record one verified fact, or abort the run.

    This raises instead of using the `assert` statement, because `python3 -O`
    removes `assert` and the checks are the whole point of the program.
    """
    global CHECKS
    if not condition:
        raise AssertionError(description)
    CHECKS += 1
    print(f"  ok: {description}")


def saga_key(booking_id: str) -> str:
    """The saga's state key. It derives from the booking id, which is also the
    partition key of the payments queue: that is what makes the payer's
    read-then-write safe, and it is stated here because it is a property of the
    naming and nothing enforces it."""
    return f"saga:{booking_id}"


async def main() -> int:
    global CHECKS
    queen = Queen(url=QUEEN_URL)

    reserve_decisions: list = []
    payments_requested: list = []
    compensations_delivered: list = []
    rooms_released: list = []
    compensations_refused: list = []
    preconditions_lost = 0
    verdict, failed = "", False

    try:
        print(f"broker {QUEEN_URL}")

        for queue in (BOOKINGS, PAYMENTS, EXPIRIES):
            await queen.queue(queue).config({"lease_time": 30, "retry_limit": 3}).create()

        # ------------------------------------------------------------ queuing
        print("\nsubmitting bookings")
        for index, booking in enumerate(BOOKINGS_IN):
            await queen.queue(BOOKINGS).push(
                {
                    # Distinct transaction ids on purpose. Deduplication would
                    # swallow the duplicate submission and the gate would never
                    # be tested, and a real redelivery arrives with an identity
                    # of its own too.
                    "transactionId": f"submit-{index}-{booking['bookingId']}",
                    "data": booking,
                }
            )
        print(f"  {len(BOOKINGS_IN)} submissions for {len(BOOKING_IDS)} bookings")

        # ---------------------------------------------------------- reserving
        #
        # The bundle, and the whole point of the example: five things commit
        # together, so there is no ordering between them left to get wrong.
        print("\nreserving")

        async def reserve(msg) -> None:
            nonlocal preconditions_lost
            booking = msg["data"]
            booking_id, room, cents = booking["bookingId"], booking["room"], booking["cents"]
            group = msg.get("consumerGroup")

            tx = queen.transaction()
            # 1. The gate AND the first state, in one row. required=True is what
            #    makes it a gate instead of a verdict: without it a lost race
            #    would come back applied=False while the payment and the timer
            #    went out anyway.
            tx.kv.put_if_absent(
                NS,
                saga_key(booking_id),
                {"step": "held", "room": room, "cents": cents},
                # The Python client takes a timedelta where the JavaScript one
                # takes "1h"; both resolve to the one field the wire has,
                # ttlSeconds.
                ttl=timedelta(hours=1),
                required=True,
            )
            # 2. The obligation. From the moment this commits it is a row in the
            #    broker's own table, so it survives this handler, this process,
            #    this deploy and this machine. The key is chosen by us, which is
            #    the entire reason it can be cancelled later by name.
            tx.timer(EXPIRIES).key(booking_id).after_ms(HOLD_MS).payload(
                {"bookingId": booking_id, "room": room}
            ).schedule()
            # 3. The work. Partitioned by booking, so every message about one
            #    booking is in one lane.
            tx.queue(PAYMENTS).partition(booking_id).push(
                {"transactionId": f"pay-{booking_id}", "data": {"bookingId": booking_id, "cents": cents}}
            )
            # 4. The acknowledgement, carrying this delivery's lease. An expired
            #    lease refuses the ack and takes the other three down with it,
            #    which is the guarantee no compare-and-swap can give.
            res = await tx.ack(msg, "completed", {"consumer_group": group}).commit()

            reserve_decisions.append(booking_id)

            # A lost gate is RETURNED, not raised: HTTP 200, success=False,
            # reason "kv_precondition". It is the ordinary outcome of every
            # legitimate redelivery, which makes it one of the most frequent
            # answers this product gives, and it does not belong in an except
            # block where the reflex is to retry.
            if res.get("success") is False and res.get("reason") == "kv_precondition":
                # Nothing was written: no second payment, no second timer, no
                # second row. The message still has to leave the cursor, so it
                # is acknowledged on its own.
                preconditions_lost += 1
                await queen.ack(msg, "completed", {"group": group})
                print(f"  {booking_id}: already held, whole bundle rolled back ({res.get('kvReason')})")
                return

            print(f"  {booking_id}: room {room} held, compensation armed for {HOLD_MS} ms")

        await (
            queen.queue(BOOKINGS)
            .group("reserver")
            .subscription_mode("all")
            .auto_ack(False)
            .each()
            # The count that ends the phase, with the deadline behind it.
            .limit(len(BOOKINGS_IN))
            .idle_millis(PHASE_MS)
            .consume(reserve)
        )

        check(
            len(reserve_decisions) == len(BOOKINGS_IN),
            f"the reserver reached a decision on every submission ({len(BOOKINGS_IN)}, got {len(reserve_decisions)})",
        )
        check(preconditions_lost == 1, "the duplicate submission lost the gate exactly once")

        # Pending timers are a table you can read, not a promise you have to
        # trust.
        armed = await queen.timers.list(EXPIRIES, limit=50)
        print(f"  timers armed: {', '.join(sorted(row['timerKey'] for row in armed['rows']))}")
        check(
            len(armed["rows"]) == len(BOOKING_IDS),
            f"one compensation is armed per booking and the duplicate added none "
            f"({len(BOOKING_IDS)}, got {len(armed['rows'])})",
        )

        # ------------------------------------------------------------- paying
        #
        # The other end of the saga. A settled payment confirms the state and
        # calls the compensation off in one commit; a declined card leaves the
        # state where it is and lets the timer do its work.
        print("\npaying")

        async def pay(msg) -> None:
            booking_id = msg["data"]["bookingId"]
            group = msg.get("consumerGroup")
            payments_requested.append(booking_id)

            # A read in one call and a write in the next. It is safe HERE
            # because the key derives from the partition key: every message
            # about this booking arrives in one lane of this queue, and a lane
            # has one reader per group. Where a key does not derive from the
            # partition key this shape is a race and the atomics are the answer,
            # which is exactly the compensator's situation further down.
            state = await queen.kv.get(NS, saga_key(booking_id))

            if booking_id == DECLINED:
                # A declined card is a business outcome, not a delivery failure:
                # the message is done with. The room stays held, and nothing in
                # this process is responsible for giving it back.
                await queen.ack(msg, "completed", {"group": group})
                print(f"  {booking_id}: card declined, hold left to expire")
                return

            tx = queen.transaction()
            # `expect` makes the serialisation assumption falsifiable instead of
            # silent. If the lane really serialises, it never fails and costs
            # nothing; the day it fails, two consumers are serving one partition
            # and you learn it as a verdict rather than as a wrong total.
            tx.kv.put(
                NS,
                saga_key(booking_id),
                {**state["value"], "step": "confirmed"},
                ttl=timedelta(hours=1),
                expect=state["version"],
                required=True,
            )

            if booking_id != CANCEL_SKIPPED:
                # The cancel rides the bundle. Either the booking is confirmed
                # and the compensation is called off, or neither happened.
                tx.timer(EXPIRIES).key(booking_id).cancel()

            res = await tx.ack(msg, "completed", {"consumer_group": group}).commit()
            if res.get("success") is False:
                raise AssertionError(f"{booking_id}: confirmation lost its fence ({res.get('kvReason')})")

            tail = (
                ", compensation deliberately NOT cancelled"
                if booking_id == CANCEL_SKIPPED
                else ", compensation cancelled"
            )
            print(f"  {booking_id}: paid and confirmed{tail}")

        await (
            queen.queue(PAYMENTS)
            .group("payer")
            .subscription_mode("all")
            .auto_ack(False)
            .each()
            .limit(len(BOOKING_IDS))
            .idle_millis(PHASE_MS)
            .consume(pay)
        )

        check(
            len(payments_requested) == len(BOOKING_IDS),
            f"every booking was asked to pay once and the duplicate produced no second payment "
            f"({len(BOOKING_IDS)}, got {len(payments_requested)})",
        )
        check(len(set(payments_requested)) == len(payments_requested), "no booking was asked to pay twice")

        # The cancel is observable before anything is delivered: the row is gone
        # from the staging table. A peek is how you ask, and a miss is
        # {"found": false} with HTTP 200, never a 404.
        peeked = {b: (await queen.timers.peek(EXPIRIES, b))["found"] for b in BOOKING_IDS}
        check(peeked["B-1"] is False, "the compensation cancelled inside the confirming bundle is gone from the table")
        check(peeked[DECLINED] is True, f"{DECLINED} was never confirmed, so its compensation is still armed")
        check(
            peeked[CANCEL_SKIPPED] is True,
            f"{CANCEL_SKIPPED} is confirmed but its compensation is still armed on purpose",
        )

        # ------------------------------------------------------- compensating
        #
        # What the timers deliver, and the consumer that must not trust them.
        #
        # A compensation message is not an instruction, it is a question: is
        # this saga still open? A fired timer leaves no tombstone, so a cancel
        # that arrives a millisecond late answers `absent` and the message goes
        # out anyway. The state is the authority and it is read first.
        #
        # And here the key does NOT derive from the partition key: this message
        # arrives on another queue entirely, in a lane that has nothing to do
        # with the payments lane, so no partitioning could serialise the two
        # writers. That is what `expect` is for, and on this path it is
        # load-bearing rather than an assertion.
        print("\ncompensating")

        async def compensate_one(msg) -> None:
            booking_id, room = msg["data"]["bookingId"], msg["data"]["room"]
            group = msg.get("consumerGroup")
            compensations_delivered.append(booking_id)

            state = await queen.kv.get(NS, saga_key(booking_id))

            if not state["found"] or state["value"]["step"] != "held":
                # The booking was confirmed before this fired. Compensating here
                # is how a saga unwinds a sale that has already shipped.
                compensations_refused.append(booking_id)
                await queen.ack(msg, "completed", {"group": group})
                step = state["value"]["step"] if state["found"] else "gone"
                print(f"  {booking_id}: state is {step}, compensation refused")
                return

            res = await (
                queen.transaction()
                .kv.put(
                    NS,
                    saga_key(booking_id),
                    {**state["value"], "step": "expired"},
                    ttl=timedelta(hours=1),
                    expect=state["version"],
                    required=True,
                )
                .ack(msg, "completed", {"consumer_group": group})
                .commit()
            )

            if res.get("success") is False:
                # Somebody confirmed it between the read and the commit. The
                # fence held, nothing was written, and the room stays sold.
                compensations_refused.append(booking_id)
                await queen.ack(msg, "completed", {"group": group})
                print(f"  {booking_id}: confirmed under us, compensation refused by the fence")
                return

            rooms_released.append(room)
            print(f"  {booking_id}: hold expired, room {room} released")

        def compensator(limit: int, idle_millis: int):
            return (
                queen.queue(EXPIRIES)
                .group("compensator")
                .subscription_mode("all")
                .auto_ack(False)
                .each()
                .limit(limit)
                .idle_millis(idle_millis)
                .consume(compensate_one)
            )

        # Two timers were left armed, so two messages must arrive: that is the
        # count, and TIMER_DEADLINE_MS is the deadline behind it.
        await compensator(2, TIMER_DEADLINE_MS)
        check(
            len(compensations_delivered) == 2,
            f"both uncancelled compensations were delivered (2, got {len(compensations_delivered)}"
            f"{': ' + ', '.join(compensations_delivered) if compensations_delivered else ''})",
        )

        # Then a bounded second pass with room for two more. It is the only
        # honest way to say "a cancelled timer never arrived": the first pass
        # would have stopped at two whatever those two were, so the claim is
        # really that nothing else shows up afterwards.
        await compensator(2, 4000)

        # ----------------------------------------------------------- checking
        print("\nchecking")

        check(
            len(compensations_delivered) == 2,
            f"nothing else arrived on a second pass: still 2 compensations (got {len(compensations_delivered)})",
        )
        check(
            "B-1" not in compensations_delivered and "B-2" not in compensations_delivered,
            "a cancelled compensation was never delivered",
        )
        check(
            rooms_released == ["103"],
            f"exactly one room went back on sale, the one whose card was declined "
            f"(got {', '.join(rooms_released) or 'none'})",
        )
        check(
            compensations_refused == [CANCEL_SKIPPED],
            "the compensation for the confirmed booking was refused by the consumer, not prevented by the cancel",
        )

        states = await queen.kv.get_many(NS, [saga_key(b) for b in BOOKING_IDS])
        check(
            len(states["rows"]) == len(BOOKING_IDS) and len(states["missing"]) == 0,
            f"every booking left exactly one saga row ({len(BOOKING_IDS)}, got {len(states['rows'])})",
        )

        step = {row["key"].replace("saga:", ""): row["value"]["step"] for row in states["rows"]}
        check(
            step["B-1"] == "confirmed" and step["B-2"] == "confirmed",
            "the two ordinary bookings ended confirmed",
        )
        check(
            step[CANCEL_SKIPPED] == "confirmed",
            f"{CANCEL_SKIPPED} is still confirmed after its compensation was delivered",
        )
        check(step[DECLINED] == "expired", f"{DECLINED} was unwound by its timer, with nobody awake to do it")

        print("\n  final: " + ", ".join(f"{k}={v}" for k, v in sorted(step.items())))

        verdict = f"\nPASS: {CHECKS} checks"
    except Exception as err:  # noqa: BLE001 - the program's verdict is its exit code
        verdict, failed = f"\nFAIL: {err}", True
    finally:
        # ---------------------------------------------------------- purge
        #
        # Three things to remove, and the first two are the ones that are easy
        # to forget. The saga rows live in their own table, and a pending timer
        # lives in the staging table keyed by NAME: neither is reached by
        # deleting the queue, and a timer whose queue no longer exists still
        # fires and provisions it again on the way out.
        #
        # Unconditional, in a finally, because a run that FAILED is exactly the
        # run whose leftovers matter: an armed timer would deliver into the next
        # run and a surviving saga row would make the next run pass without
        # holding anything.
        #
        # Best effort: a purge that raised would replace the real verdict with
        # its own.
        try:
            for booking_id in BOOKING_IDS:
                await queen.timers.cancel(EXPIRIES, booking_id)
                await queen.kv.delete(NS, saga_key(booking_id))
            for queue in (BOOKINGS, PAYMENTS, EXPIRIES):
                await queen.queue(queue).delete()
        except Exception as err:  # noqa: BLE001 - the run's verdict outranks this
            print(f"  (purge incomplete: {err})")
        await queen.close()

    sys.stdout.flush()
    print(verdict, file=sys.stderr if failed else sys.stdout)
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))
```
### Go

```go title="examples/apps/go/saga/main.go"
//
// A booking saga whose compensation is a timer, and whose every step is a row
// in the same PostgreSQL as the queue.
//
// The war story is a room hold that never came back. A booking system held
// inventory when a reservation started and released it when the payment either
// settled or failed, and the release lived in a time.AfterFunc inside the
// worker. A rolling deploy replaced the workers; every hold in flight lost its
// release; and a fortnight later somebody noticed a hotel had been sold out on
// paper for nine nights it had spent empty.
//
// The release was not slow, it was in the wrong place. A compensation is not a
// timeout, it is an obligation, and an obligation has to outlive the process
// that took it on. Here the gate, the saga state, the compensation timer, the
// payment request and the acknowledgement are ONE PostgreSQL transaction. If
// the room is held, the compensation exists. If the room is not held, nothing
// else happened either. There is no interval in between for a deploy to land
// in.
//
//	bookings
//	  |-- group "reserver"   ONE bundle: gate + state + timer + push + ack
//	        |-- payments (partitioned by booking)
//	        |     `-- group "payer"    confirm + CANCEL the timer + ack, one bundle
//	        `-- expiries (delivered by the timer, at the hold's expiry)
//	              `-- group "compensator"   reads the state BEFORE compensating
//
// Run it:
//
//	QUEEN_URL=http://localhost:6632 GOWORK=off go run ./saga
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"

	queen "github.com/smartpricing/queen/clients/client-go"
)

// A suffix on the queue names AND on the KV namespace. The queues need it
// because delete-then-recreate leaves stale partition state for up to 30
// seconds; the namespace needs it for the opposite reason -- a saga row
// outlives the run that wrote it, so a second run under the same namespace
// would find every booking already held and measure nothing.
var runID = strconv.FormatInt(time.Now().UnixMilli(), 36)

var (
	bookingsQueue = "app-go-saga-bookings-" + runID
	paymentsQueue = "app-go-saga-payments-" + runID
	expiriesQueue = "app-go-saga-expiries-" + runID
	ns            = "app-go-saga-" + runID
)

const (
	reserverGroup    = "reserver"
	payerGroup       = "payer"
	compensatorGroup = "compensator"

	// How long a room stays held before the compensation fires. Short enough
	// for a test, and in production it is the only number that changes.
	//
	// It has to outlast the reserve and pay phases, or a timer would fire
	// before the payment that cancels it and the run would be measuring a race
	// rather than a design. deliverAt is a floor and never a ceiling, so a
	// timer can only be late: a margin here is sound, where a margin in the
	// other direction would not be.
	hold = 15 * time.Second

	// Every phase below ends on a COUNT, and this is the deadline behind the
	// count. Never wait for silence: a phase that stops when nothing has
	// arrived for a while passes on a broker that delivered nothing at all.
	// Reaching the deadline ends the phase short, and the count check that
	// follows is what reports it.
	phaseMillis = 20000

	// The compensation phase gets its own, longer deadline. A timer fires no
	// earlier than its delay plus one sweeper cycle, and a broker whose timer
	// table has been empty for a while wakes up lazily.
	timerDeadlineMillis = 90000

	// B-3's card is declined, so its saga never reaches "confirmed" and the
	// timer is the thing that gives the room back. It is the compensation
	// actually doing its job.
	declined = "B-3"

	// B-4 pays, but its cancel is deliberately skipped, which makes the race
	// deterministic: a cancel that arrives after the fire answers `absent`, and
	// ABSENT MAY MEAN ALREADY DELIVERED. So the compensation for a confirmed
	// booking has to be refused by the consumer that receives it, never
	// prevented by the cancel alone.
	cancelSkipped = "B-4"
)

// Four bookings, five submissions. B-2 is submitted twice: the same booking,
// two messages, which is what a redelivery looks like from the reserver's side
// and the reason the bundle opens with a gate rather than with a check.
type booking struct {
	BookingID string `json:"bookingId"`
	Room      string `json:"room"`
	Cents     int    `json:"cents"`
}

var bookingsIn = []booking{
	{BookingID: "B-1", Room: "101", Cents: 24000},
	{BookingID: "B-2", Room: "102", Cents: 31000},
	{BookingID: "B-2", Room: "102", Cents: 31000},
	{BookingID: "B-3", Room: "103", Cents: 18000},
	{BookingID: "B-4", Room: "104", Cents: 27000},
}

var bookingIDs = []string{"B-1", "B-2", "B-3", "B-4"}

// sagaState is what the KV row holds. It is a struct rather than a map because
// the value comes back as raw JSON and this program reads a field of it on
// every hop: a typo in a map key would read as "the saga is not held" and the
// run would pass for the wrong reason.
type sagaState struct {
	Step  string `json:"step"`
	Room  string `json:"room"`
	Cents int    `json:"cents"`
}

// The saga's state key. It derives from the booking id, which is also the
// partition key of the payments queue: that is what makes the payer's
// read-then-write safe, and it is stated here because it is a property of the
// naming and nothing enforces it.
func sagaKey(bookingID string) string { return "saga:" + bookingID }

var checks int

// assert is the whole test framework here. Go has no exceptions, so a failed
// check is an error that unwinds run() and is printed once, at the bottom.
func assert(condition bool, description string) error {
	if !condition {
		return fmt.Errorf("%s", description)
	}
	checks++
	fmt.Printf("  ok: %s\n", description)
	return nil
}

func main() {
	if err := run(); err != nil {
		fmt.Fprintf(os.Stderr, "\nFAIL: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("\nPASS: %d checks\n", checks)
}

func run() error {
	brokerURL := os.Getenv("QUEEN_URL")
	if brokerURL == "" {
		brokerURL = "http://localhost:6632"
	}

	// Every call in the Go client takes a context, and it is the only deadline
	// there is. This one bounds the whole program, so a broker that stops
	// answering ends the run instead of wedging it.
	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second)
	defer cancel()

	client, err := queen.New(brokerURL)
	if err != nil {
		return fmt.Errorf("create client: %w", err)
	}
	defer client.Close(context.Background())

	// ------------------------------------------------------------------ purge
	//
	// Three things to remove, and the first two are the ones that are easy to
	// forget. The saga rows live in their own table, and a pending timer lives
	// in the staging table keyed by NAME: neither is reached by deleting the
	// queue, and a timer whose queue no longer exists still fires and
	// provisions it again on the way out.
	//
	// Deferred rather than run at the end, because a run that FAILED is
	// exactly the run whose leftovers matter: an armed timer would deliver
	// into the next run and a surviving saga row would make the next run pass
	// without holding anything.
	//
	// Best effort throughout, on its own context: a purge that reported its
	// own trouble as the verdict would hide the real one.
	defer func() {
		purgeCtx, purgeCancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer purgeCancel()
		for _, bookingID := range bookingIDs {
			if _, err := client.Timers().Cancel(purgeCtx, expiriesQueue, bookingID); err != nil {
				fmt.Fprintf(os.Stderr, "  (purge incomplete: cancel %s: %v)\n", bookingID, err)
			}
			if _, err := client.KV().Delete(purgeCtx, ns, sagaKey(bookingID)); err != nil {
				fmt.Fprintf(os.Stderr, "  (purge incomplete: delete %s: %v)\n", sagaKey(bookingID), err)
			}
		}
		for _, q := range []string{bookingsQueue, paymentsQueue, expiriesQueue} {
			if _, err := client.Queue(q).Delete().Execute(purgeCtx); err != nil {
				fmt.Fprintf(os.Stderr, "  (purge incomplete: delete %s: %v)\n", q, err)
			}
		}
	}()

	fmt.Printf("broker %s\n", brokerURL)

	for _, q := range []string{bookingsQueue, paymentsQueue, expiriesQueue} {
		if _, err := client.Queue(q).
			Config(queen.QueueConfig{LeaseTime: 30, RetryLimit: 3}).
			Create().Execute(ctx); err != nil {
			return fmt.Errorf("create %s: %w", q, err)
		}
	}

	// ---------------------------------------------------------------- queuing
	fmt.Println("\nsubmitting bookings")
	for i, b := range bookingsIn {
		if _, err := client.Queue(bookingsQueue).
			Push(b).
			// Distinct transaction ids on purpose. Deduplication would swallow
			// the duplicate submission and the gate would never be tested, and
			// a real redelivery arrives with an identity of its own too.
			TransactionID(fmt.Sprintf("submit-%d-%s", i, b.BookingID)).
			Execute(ctx); err != nil {
			return fmt.Errorf("submit %s: %w", b.BookingID, err)
		}
	}
	fmt.Printf("  %d submissions for %d bookings\n", len(bookingsIn), len(bookingIDs))

	// -------------------------------------------------------------- reserving
	//
	// The bundle, and the whole point of the example: five things commit
	// together, so there is no ordering between them left to get wrong.
	//
	// Concurrency is the default of one, so this handler runs on a single
	// goroutine and the counters below need no lock.
	fmt.Println("\nreserving")
	var reserveDecisions []string
	preconditionsLost := 0

	err = client.Queue(bookingsQueue).
		Group(reserverGroup).
		SubscriptionMode(queen.SubscriptionModeAll).
		AutoAck(false).
		Each().
		// The count that ends the phase, with the deadline behind it.
		Limit(len(bookingsIn)).
		IdleMillis(phaseMillis).
		// Each poll is capped at a second so the idle deadline is noticed
		// promptly rather than inside a 30 s long poll.
		TimeoutMillis(1000).
		Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
			bookingID, _ := msg.Data["bookingId"].(string)
			room, _ := msg.Data["room"].(string)
			cents, ok := msg.Data["cents"].(float64)
			if !ok {
				return fmt.Errorf("submission %s carries no numeric cents", msg.TransactionID)
			}

			res, err := client.Transaction().
				// 1. The gate AND the first state, in one row. Required is what
				//    makes it a gate instead of a verdict: without it a lost
				//    race would come back applied:false while the payment and
				//    the timer went out anyway.
				KV(queen.KVPutIfAbsentOp(
					ns,
					sagaKey(bookingID),
					sagaState{Step: "held", Room: room, Cents: int(cents)},
					// An expiry is mandatory on every KV write, and the zero
					// value of queen.Expiry is refused rather than treated as
					// "no opinion". Forever exists and is never used here: an
					// example that runs in CI must not be able to leave an
					// immortal row behind.
					queen.TTL(time.Hour),
					queen.KVWriteOptions{Required: true},
				)).
				// 2. The obligation. From the moment this commits it is a row
				//    in the broker's own table, so it survives this handler,
				//    this process, this deploy and this machine. The key is
				//    chosen by us, which is the entire reason it can be
				//    cancelled later by name.
				Timers(queen.ScheduleTimerOp(queen.TimerSchedule{
					Queue:    expiriesQueue,
					TimerKey: bookingID,
					Delay:    hold,
					Payload:  map[string]interface{}{"bookingId": bookingID, "room": room},
				})).
				// 3. The work. Partitioned by booking, so every message about
				//    one booking is in one lane.
				Queue(paymentsQueue).
				Partition(bookingID).
				Push(queen.PushItem{
					TransactionID: "pay-" + bookingID,
					Payload:       map[string]interface{}{"bookingId": bookingID, "cents": int(cents)},
				}).
				// 4. The acknowledgement, carrying this delivery's lease. An
				//    expired lease refuses the ack and takes the other three
				//    down with it, which is the guarantee that no
				//    compare-and-swap can give.
				Ack(msg, "completed", queen.AckOptions{ConsumerGroup: reserverGroup}).
				Commit(ctx)

			reserveDecisions = append(reserveDecisions, bookingID)

			// A lost gate is RETURNED, not an error: HTTP 200, Success false,
			// Reason "kv_precondition". It is the ordinary outcome of every
			// legitimate redelivery, which makes it one of the most frequent
			// answers this product gives, and it must stay out of the error
			// path where the reflex is to retry. Every OTHER failure of a
			// commit IS an error, and this handler returns it.
			if res.IsKVPrecondition() {
				// Nothing was written: no second payment, no second timer, no
				// second row. The message still has to leave the cursor, so it
				// is acknowledged on its own.
				preconditionsLost++
				if _, err := client.Ack(ctx, msg, true, queen.AckOptions{ConsumerGroup: reserverGroup}); err != nil {
					return fmt.Errorf("ack the duplicate submission: %w", err)
				}
				fmt.Printf("  %s: already held, whole bundle rolled back (%s)\n", bookingID, res.KVReason)
				return nil
			}
			if err != nil {
				return fmt.Errorf("reserve %s: %w", bookingID, err)
			}

			fmt.Printf("  %s: room %s held, compensation armed for %s\n", bookingID, room, hold)
			return nil
		}).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("reserving: %w", err)
	}

	if err := assert(
		len(reserveDecisions) == len(bookingsIn),
		fmt.Sprintf("the reserver reached a decision on every submission (%d, got %d)", len(bookingsIn), len(reserveDecisions)),
	); err != nil {
		return err
	}
	if err := assert(preconditionsLost == 1, "the duplicate submission lost the gate exactly once"); err != nil {
		return err
	}

	// Pending timers are a table you can read, not a promise you have to trust.
	armed, err := client.Timers().List(ctx, expiriesQueue, queen.TimerListOptions{Limit: 50})
	if err != nil {
		return fmt.Errorf("list the armed compensations: %w", err)
	}
	armedKeys := make([]string, 0, len(armed.Rows))
	for _, row := range armed.Rows {
		armedKeys = append(armedKeys, row.TimerKey)
	}
	sort.Strings(armedKeys)
	fmt.Printf("  timers armed: %s\n", strings.Join(armedKeys, ", "))
	if err := assert(
		len(armedKeys) == len(bookingIDs),
		fmt.Sprintf("one compensation is armed per booking and the duplicate added none (%d, got %d)", len(bookingIDs), len(armedKeys)),
	); err != nil {
		return err
	}

	// ----------------------------------------------------------------- paying
	//
	// The other end of the saga. A settled payment confirms the state and calls
	// the compensation off in one commit; a declined card leaves the state
	// where it is and lets the timer do its work.
	fmt.Println("\npaying")
	var paymentsRequested []string

	err = client.Queue(paymentsQueue).
		Group(payerGroup).
		SubscriptionMode(queen.SubscriptionModeAll).
		AutoAck(false).
		Each().
		Limit(len(bookingIDs)).
		IdleMillis(phaseMillis).
		TimeoutMillis(1000).
		// Each key's payments land in that key's partition, and a pop claims a
		// single partition unless it is asked for more.
		Partitions(10).
		Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
			bookingID, _ := msg.Data["bookingId"].(string)
			paymentsRequested = append(paymentsRequested, bookingID)

			// A read in one call and a write in the next. It is safe HERE
			// because the key derives from the partition key: every message
			// about this booking arrives in one lane of this queue, and a lane
			// has one reader per group. Where a key does not derive from the
			// partition key, this shape is a race and the atomics are the
			// answer -- which is exactly the compensator's situation below.
			state, version, err := readSaga(ctx, client, bookingID)
			if err != nil {
				return err
			}

			if bookingID == declined {
				// A declined card is a business outcome, not a delivery
				// failure: the message is done with. The room stays held, and
				// nothing in this process is responsible for giving it back.
				if _, err := client.Ack(ctx, msg, true, queen.AckOptions{ConsumerGroup: payerGroup}); err != nil {
					return fmt.Errorf("ack the declined payment: %w", err)
				}
				fmt.Printf("  %s: card declined, hold left to expire\n", bookingID)
				return nil
			}

			state.Step = "confirmed"
			tx := client.Transaction().
				// Expect makes the serialisation assumption falsifiable
				// instead of silent. If the lane really serialises, it never
				// fails and costs nothing; the day it fails, two consumers are
				// serving one partition and you learn it as a verdict rather
				// than as a wrong total.
				KV(queen.KVPutOp(ns, sagaKey(bookingID), state, queen.TTL(time.Hour), queen.KVWriteOptions{
					Expect:   queen.Expect(version),
					Required: true,
				}))

			if bookingID != cancelSkipped {
				// The cancel rides the bundle. Either the booking is confirmed
				// and the compensation is called off, or neither of the two
				// happened.
				tx = tx.Timers(queen.CancelTimerOp(expiriesQueue, bookingID))
			}

			res, err := tx.Ack(msg, "completed", queen.AckOptions{ConsumerGroup: payerGroup}).Commit(ctx)
			if res.IsKVPrecondition() {
				return fmt.Errorf("%s: confirmation lost its fence (%s)", bookingID, res.KVReason)
			}
			if err != nil {
				return fmt.Errorf("confirm %s: %w", bookingID, err)
			}

			tail := ", compensation cancelled"
			if bookingID == cancelSkipped {
				tail = ", compensation deliberately NOT cancelled"
			}
			fmt.Printf("  %s: paid and confirmed%s\n", bookingID, tail)
			return nil
		}).
		Execute(ctx)
	if err != nil {
		return fmt.Errorf("paying: %w", err)
	}

	if err := assert(
		len(paymentsRequested) == len(bookingIDs),
		fmt.Sprintf("every booking was asked to pay once and the duplicate produced no second payment (%d, got %d)",
			len(bookingIDs), len(paymentsRequested)),
	); err != nil {
		return err
	}
	if err := assert(distinct(paymentsRequested), "no booking was asked to pay twice"); err != nil {
		return err
	}

	// The cancel is observable before anything is delivered: the row is gone
	// from the staging table. A peek is how you ask, and a miss is
	// Found:false with HTTP 200, never a 404.
	peeked := map[string]bool{}
	for _, bookingID := range bookingIDs {
		info, err := client.Timers().Peek(ctx, expiriesQueue, bookingID)
		if err != nil {
			return fmt.Errorf("peek %s: %w", bookingID, err)
		}
		peeked[bookingID] = info.Found
	}
	if err := assert(!peeked["B-1"], "the compensation cancelled inside the confirming bundle is gone from the table"); err != nil {
		return err
	}
	if err := assert(peeked[declined], declined+" was never confirmed, so its compensation is still armed"); err != nil {
		return err
	}
	if err := assert(peeked[cancelSkipped], cancelSkipped+" is confirmed but its compensation is still armed on purpose"); err != nil {
		return err
	}

	// ------------------------------------------------------------ compensating
	//
	// What the timers deliver, and the consumer that must not trust them.
	//
	// A compensation message is not an instruction, it is a question: is this
	// saga still open? A fired timer leaves no tombstone, so a cancel that
	// arrives a millisecond late answers `absent` and the message goes out
	// anyway. The state is the authority and it is read first.
	//
	// And here the key does NOT derive from the partition key: this message
	// arrives on another queue entirely, in a lane that has nothing to do with
	// the payments lane, so no partitioning could serialise the two writers.
	// That is what Expect is for, and on this path it is load-bearing rather
	// than an assertion.
	fmt.Println("\ncompensating")
	var compensationsDelivered []string
	var roomsReleased []string
	var compensationsRefused []string

	compensate := func(limit, idleMillis int) error {
		return client.Queue(expiriesQueue).
			Group(compensatorGroup).
			SubscriptionMode(queen.SubscriptionModeAll).
			AutoAck(false).
			Each().
			Limit(limit).
			IdleMillis(idleMillis).
			TimeoutMillis(1000).
			Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
				bookingID, _ := msg.Data["bookingId"].(string)
				room, _ := msg.Data["room"].(string)
				compensationsDelivered = append(compensationsDelivered, bookingID)

				state, version, err := readSaga(ctx, client, bookingID)
				if err != nil {
					return err
				}

				if state.Step != "held" {
					// The booking was confirmed before this fired.
					// Compensating here is how a saga unwinds a sale that has
					// already shipped.
					compensationsRefused = append(compensationsRefused, bookingID)
					if _, err := client.Ack(ctx, msg, true, queen.AckOptions{ConsumerGroup: compensatorGroup}); err != nil {
						return fmt.Errorf("ack the refused compensation: %w", err)
					}
					step := state.Step
					if step == "" {
						step = "gone"
					}
					fmt.Printf("  %s: state is %s, compensation refused\n", bookingID, step)
					return nil
				}

				state.Step = "expired"
				res, err := client.Transaction().
					KV(queen.KVPutOp(ns, sagaKey(bookingID), state, queen.TTL(time.Hour), queen.KVWriteOptions{
						Expect:   queen.Expect(version),
						Required: true,
					})).
					Ack(msg, "completed", queen.AckOptions{ConsumerGroup: compensatorGroup}).
					Commit(ctx)

				if res.IsKVPrecondition() {
					// Somebody confirmed it between the read and the commit.
					// The fence held, nothing was written, and the room stays
					// sold.
					compensationsRefused = append(compensationsRefused, bookingID)
					if _, err := client.Ack(ctx, msg, true, queen.AckOptions{ConsumerGroup: compensatorGroup}); err != nil {
						return fmt.Errorf("ack the fenced compensation: %w", err)
					}
					fmt.Printf("  %s: confirmed under us, compensation refused by the fence\n", bookingID)
					return nil
				}
				if err != nil {
					return fmt.Errorf("compensate %s: %w", bookingID, err)
				}

				roomsReleased = append(roomsReleased, room)
				fmt.Printf("  %s: hold expired, room %s released\n", bookingID, room)
				return nil
			}).
			Execute(ctx)
	}

	// Two timers were left armed, so two messages must arrive: that is the
	// count, and timerDeadlineMillis is the deadline behind it.
	if err := compensate(2, timerDeadlineMillis); err != nil {
		return fmt.Errorf("compensating: %w", err)
	}
	if err := assert(
		len(compensationsDelivered) == 2,
		fmt.Sprintf("both uncancelled compensations were delivered (2, got %d: %s)",
			len(compensationsDelivered), strings.Join(compensationsDelivered, ", ")),
	); err != nil {
		return err
	}

	// Then a bounded second pass with room for two more. It is the only honest
	// way to say "a cancelled timer never arrived": the first pass would have
	// stopped at two whatever those two were, so the claim is really that
	// nothing else shows up afterwards.
	if err := compensate(2, 4000); err != nil {
		return fmt.Errorf("second compensation pass: %w", err)
	}

	// --------------------------------------------------------------- checking
	fmt.Println("\nchecking")

	if err := assert(
		len(compensationsDelivered) == 2,
		fmt.Sprintf("nothing else arrived on a second pass: still 2 compensations (got %d)", len(compensationsDelivered)),
	); err != nil {
		return err
	}
	if err := assert(
		!contains(compensationsDelivered, "B-1") && !contains(compensationsDelivered, "B-2"),
		"a cancelled compensation was never delivered",
	); err != nil {
		return err
	}
	if err := assert(
		len(roomsReleased) == 1 && roomsReleased[0] == "103",
		fmt.Sprintf("exactly one room went back on sale, the one whose card was declined (got %s)",
			orNone(strings.Join(roomsReleased, ", "))),
	); err != nil {
		return err
	}
	if err := assert(
		len(compensationsRefused) == 1 && compensationsRefused[0] == cancelSkipped,
		"the compensation for the confirmed booking was refused by the consumer, not prevented by the cancel",
	); err != nil {
		return err
	}

	keys := make([]string, 0, len(bookingIDs))
	for _, bookingID := range bookingIDs {
		keys = append(keys, sagaKey(bookingID))
	}
	states, err := client.KV().GetMany(ctx, ns, keys)
	if err != nil {
		return fmt.Errorf("read the saga rows: %w", err)
	}
	if err := assert(
		len(states.Rows) == len(bookingIDs) && len(states.Missing) == 0,
		fmt.Sprintf("every booking left exactly one saga row (%d, got %d)", len(bookingIDs), len(states.Rows)),
	); err != nil {
		return err
	}

	step := map[string]string{}
	for _, row := range states.Rows {
		var st sagaState
		if err := json.Unmarshal(row.Value, &st); err != nil {
			return fmt.Errorf("decode saga row %s: %w", row.Key, err)
		}
		step[strings.TrimPrefix(row.Key, "saga:")] = st.Step
	}

	if err := assert(
		step["B-1"] == "confirmed" && step["B-2"] == "confirmed",
		"the two ordinary bookings ended confirmed",
	); err != nil {
		return err
	}
	if err := assert(
		step[cancelSkipped] == "confirmed",
		cancelSkipped+" is still confirmed after its compensation was delivered",
	); err != nil {
		return err
	}
	if err := assert(
		step[declined] == "expired",
		declined+" was unwound by its timer, with nobody awake to do it",
	); err != nil {
		return err
	}

	final := make([]string, 0, len(step))
	for k, v := range step {
		final = append(final, k+"="+v)
	}
	sort.Strings(final)
	fmt.Printf("\n  final: %s\n", strings.Join(final, ", "))

	return nil
}

// readSaga reads one saga row and its version. The version is what a later
// write passes back as Expect, so the two always travel together.
//
// A key past its expiry is never returned and never counts as existing, even
// while the sweeper has not pruned it: an absent row comes back as the zero
// sagaState, whose Step is the empty string and is therefore never "held".
func readSaga(ctx context.Context, client *queen.Queen, bookingID string) (sagaState, int64, error) {
	var state sagaState
	entry, err := client.KV().Get(ctx, ns, sagaKey(bookingID))
	if err != nil {
		return state, 0, fmt.Errorf("read the saga state of %s: %w", bookingID, err)
	}
	if !entry.Found {
		return state, 0, nil
	}
	if err := json.Unmarshal(entry.Value, &state); err != nil {
		return state, 0, fmt.Errorf("decode the saga state of %s: %w", bookingID, err)
	}
	return state, entry.Version, nil
}

func contains(values []string, want string) bool {
	for _, v := range values {
		if v == want {
			return true
		}
	}
	return false
}

func distinct(values []string) bool {
	seen := map[string]bool{}
	for _, v := range values {
		if seen[v] {
			return false
		}
		seen[v] = true
	}
	return true
}

func orNone(s string) string {
	if s == "" {
		return "none"
	}
	return s
}
```
### HTTP

```bash title="examples/apps/http/saga.sh"
#!/usr/bin/env bash
#
# A booking saga whose compensation is a timer, with nothing but curl.
#
# The war story is a room hold that never came back. A booking system held
# inventory when a reservation started and released it when the payment either
# settled or failed, and the release lived in a sleeping worker. A rolling
# deploy replaced the workers; every hold in flight lost its release; and a
# fortnight later somebody noticed a hotel had been sold out on paper for nine
# nights it had spent empty.
#
# The release was not slow, it was in the wrong place. A compensation is not a
# timeout, it is an obligation, and an obligation has to outlive the process
# that took it on. Here the gate, the saga state, the compensation timer, the
# payment request and the acknowledgement are ONE PostgreSQL transaction. If the
# room is held, the compensation exists. If the room is not held, nothing else
# happened either.
#
#   bookings
#     |-- group "reserver"    ONE bundle: gate + state + timer + push + ack
#     |     `-- payments (partitioned by booking)
#     |           `-- group "payer"   confirm + CANCEL the timer + ack, one bundle
#     `-- expiries (delivered by the timer, at the hold's expiry)
#           `-- group "compensator"   reads the state BEFORE compensating
#
# There is no client library here and none is needed, and this file is worth
# reading even if you use one. `kv` and `timers` are keys of the ROOT of the
# transaction request, beside `operations` and never elements of it, and a timer
# payload travels base64. Everything an SDK hides is written out.
#
# Run it:
#   QUEEN_URL=http://localhost:6632 bash saga.sh

set -euo pipefail

QUEEN_URL="${QUEEN_URL:-http://localhost:6632}"

# A suffix on the queue names AND on the KV namespace. The queues need it
# because delete-then-recreate leaves stale partition state for up to 30
# seconds; the namespace needs it for the opposite reason -- a saga row outlives
# the run that wrote it, so a second run under the same namespace would find
# every booking already held and measure nothing. $$ is the process id, which
# keeps two runs in the same second apart.
RUN="$(date +%s)-$$"
BOOKINGS="app-http-saga-bookings-$RUN"
PAYMENTS="app-http-saga-payments-$RUN"
EXPIRIES="app-http-saga-expiries-$RUN"
NS="app-http-saga-$RUN"

# A group's cursor lives on the queue, and the queue names are already unique
# per run, so these need no suffix.
RESERVER=app-http-reserver
PAYER=app-http-payer
COMPENSATOR=app-http-compensator

# Four bookings, five submissions. B-2 is submitted twice: the same booking, two
# messages, which is what a redelivery looks like from the reserver's side and
# the reason the bundle opens with a gate rather than with a check.
BOOKING_IDS="B-1 B-2 B-3 B-4"
SUBMISSIONS="B-1 B-2 B-2 B-3 B-4"
SUBMISSION_COUNT=5
BOOKING_COUNT=4

# B-3's card is declined, so its saga never reaches "confirmed" and the timer is
# the thing that gives the room back.
DECLINED=B-3

# B-4 pays, but its cancel is deliberately skipped, which makes the race
# deterministic: a cancel that arrives after the fire answers `absent`, and
# ABSENT MAY MEAN ALREADY DELIVERED. So the compensation for a confirmed booking
# has to be refused by the consumer that receives it, never prevented by the
# cancel alone.
CANCEL_SKIPPED=B-4

# How long a room stays held before the compensation fires. It has to outlast
# the reserve and pay phases, or a timer would fire before the payment that
# cancels it and the run would be measuring a race rather than a design.
# deliverAt is a floor and never a ceiling, so a timer can only be late: a
# margin here is sound, where a margin the other way would not be.
HOLD_MS=15000

# Every phase ends on a COUNT, and these are the deadlines behind the counts.
# Never wait for silence; wait for a total, with a deadline. The compensation
# phase gets a longer one: a timer fires no earlier than its delay plus one
# sweeper cycle, and a broker whose timer table has been empty for a while wakes
# up lazily.
PHASE_MS=20000
TIMER_DEADLINE_MS=90000

# Every pop long-polls for this many milliseconds and no longer.
POLL_MS=1000

command -v jq >/dev/null 2>&1 || { echo "FAIL: jq is not installed"; exit 1; }

CHECKS=0
TMP="$(mktemp -d)"

# One line per delivery handled: "<group> <booking> <outcome>".
OBSERVED="$TMP/observed"
: > "$OBSERVED"

# One line per room actually put back on sale. This is the compensation's
# external effect, and the whole reason the program exists.
RELEASED="$TMP/released"
: > "$RELEASED"

fail() { FAILURE="$*"; exit 1; }

# check <actual> <expected> <description>
check() {
  [ "$1" = "$2" ] || fail "$3 (expected [$2], got [$1])"
  CHECKS=$((CHECKS + 1))
  echo "  ok: $3"
}

# ok <description>: records a check whose condition was already tested, for the
# assertions that are not an equality.
ok() {
  CHECKS=$((CHECKS + 1))
  echo "  ok: $1"
}

# A millisecond clock. GNU date spells it %3N; BSD date (macOS) has no %N and
# leaves the unconverted tail in the output, so a probe for anything that is not
# a digit tells the two apart, and perl, whose Time::HiRes is core, is the
# fallback.
if [ -z "$(date +%s%3N 2>/dev/null | tr -d '0-9')" ]; then
  now_ms() { date +%s%3N; }
else
  command -v perl >/dev/null 2>&1 \
    || { echo "FAIL: need GNU date or perl for a millisecond clock"; exit 1; }
  now_ms() { perl -MTime::HiRes -e 'printf "%d", Time::HiRes::time() * 1000'; }
fi

# Sets $STATUS to the HTTP status code and writes the response body to $OUT.
# There is no --fail: Queen reports outcomes in the body and several of the
# interesting ones arrive as 200, so read the status, then the body.
OUT="$TMP/body"
request() {
  local method="$1" path="$2" body="${3:-}"
  if [ -n "$body" ]; then
    STATUS="$(curl -sS -o "$OUT" -w '%{http_code}' \
      -X "$method" "$QUEEN_URL$path" \
      -H 'content-type: application/json' -d "$body")"
  else
    STATUS="$(curl -sS -o "$OUT" -w '%{http_code}' -X "$method" "$QUEEN_URL$path")"
  fi
}

# saga_key <booking>: the state key. It derives from the booking id, which is
# also the partition key of the payments queue, and that derivation is what
# makes the payer's read-then-write safe further down.
saga_key() { printf 'saga:%s' "$1"; }

# kv_get <key>: prints the whole row as JSON, {"found":false,...} when absent.
# `found` is a field of its own because null is a legal stored value: absence is
# never inferred from the value being empty. Everything goes through
# POST /api/v1/kv, including single-key reads, which keeps the key out of access
# logs, proxy samples and tracing spans.
kv_get() {
  local body
  body="$(jq -cn --arg ns "$NS" --arg key "$1" \
    '{operations: [{op: "get", ns: $ns, key: $key}]}')"
  request POST /api/v1/kv "$body"
  [ "$STATUS" = 200 ] || fail "kv get returned HTTP $STATUS"
  jq -c '.results[0]' "$OUT"
}

# One exit path for everything. A failed check calls fail(), which records the
# reason and exits 1; any other command that fails under `set -e` arrives here
# too, with its own status. FAIL is printed exactly once, and only on failure.
#
# It also purges, and there are three things to remove. Two of them are the ones
# that are easy to forget: the saga rows live in their own table, and a PENDING
# TIMER lives in the staging table keyed by NAME, so neither is reached by
# deleting the queue -- and a timer whose queue no longer exists still fires and
# provisions the queue again on the way out.
#
# The purge is UNCONDITIONAL, because a run that failed is exactly the run whose
# leftovers matter: an armed timer would deliver into the next run, and a
# surviving saga row would make the next run pass without holding anything. And
# it is best effort, with `|| true` throughout, so a purge that fails cannot
# overwrite the verdict.
cleanup() {
  local status=$?
  purge || true
  rm -rf "$TMP"
  if [ "$status" -ne 0 ]; then
    echo
    echo "FAIL: ${FAILURE:-a command exited with status $status}"
  fi
  exit "$status"
}

purge() {
  local booking keys body
  for booking in $BOOKING_IDS; do
    # The cancel route, DELETE /api/v1/timers/:queue/*timerKey. It is the one
    # route a proxy may never block, because the fire never switches itself off.
    request DELETE "/api/v1/timers/$EXPIRIES/$booking" || true
  done
  keys="$(printf '%s\n' $BOOKING_IDS | jq -R 'sub("^"; "saga:")' | jq -sc .)" || return 0
  body="$(jq -cn --arg ns "$NS" --argjson keys "$keys" \
    '{operations: [$keys[] | {op: "delete", ns: $ns, key: .}]}')" || return 0
  request POST /api/v1/kv "$body" || true
  request DELETE "/api/v1/resources/queues/$BOOKINGS" || true
  request DELETE "/api/v1/resources/queues/$PAYMENTS" || true
  request DELETE "/api/v1/resources/queues/$EXPIRIES" || true
}
trap cleanup EXIT

echo "broker $QUEEN_URL"

# Every broker serves /api/v1/kv and /api/v1/timers: there is no flag that turns
# them on. What can still refuse is an operator's runtime kill switch (503) or a
# quota (403), so probe once here and name that, rather than letting the first
# real call fail with something that reads like a bug.
request POST /api/v1/kv '{"operations":[{"op":"get","ns":"probe","key":"probe"}]}'
[ "$STATUS" = 200 ] \
  || fail "the kv probe returned HTTP $STATUS: $(cat "$OUT") (503 is an operator's kill switch, 403 a quota; see /deploy/state)"
request GET "/api/v1/timers/$EXPIRIES?limit=1"
[ "$STATUS" = 200 ] \
  || fail "the timers probe returned HTTP $STATUS: $(cat "$OUT") (503 is an operator's kill switch, 403 a quota; see /deploy/state)"

# /configure is a full replace rather than a patch, so what is not named here is
# reset to its default.
for queue in "$BOOKINGS" "$PAYMENTS" "$EXPIRIES"; do
  body="$(jq -n --arg queue "$queue" '{queue: $queue, options: {leaseTime: 30, retryLimit: 3}}')"
  request POST /api/v1/configure "$body"
  [ "$STATUS" = 200 ] || fail "configure of $queue returned HTTP $STATUS"
done
check "$(jq -r .configured "$OUT")" true 'three queues exist, each with a 30 second lease'

# ---------------------------------------------------------------------- queuing
echo
echo "submitting bookings"
index=0
room=101
cents=24000
for booking in $SUBMISSIONS; do
  # Distinct transaction ids on purpose. Deduplication would swallow the
  # duplicate submission and the gate would never be tested, and a real
  # redelivery arrives with an identity of its own too. The duplicate carries
  # the same room and price, being the same booking submitted twice.
  case "$booking" in
    B-1) room=101; cents=24000 ;;
    B-2) room=102; cents=31000 ;;
    B-3) room=103; cents=18000 ;;
    B-4) room=104; cents=27000 ;;
  esac
  body="$(jq -n --arg queue "$BOOKINGS" --arg booking "$booking" --arg room "$room" \
    --argjson cents "$cents" --argjson i "$index" \
    '{items: [{queue: $queue, transactionId: ("submit-" + ($i|tostring) + "-" + $booking),
               payload: {bookingId: $booking, room: $room, cents: $cents}}]}')"
  request POST /api/v1/push "$body"
  [ "$STATUS" = 201 ] || fail "push of $booking returned HTTP $STATUS"
  # HTTP 201 is not proof the message was stored: "buffered" and "failed" also
  # come back 201. The per-item status is the only answer.
  [ "$(jq -r '.[0].status' "$OUT")" = queued ] \
    || fail "push of $booking came back $(jq -r '.[0].status' "$OUT")"
  index=$((index + 1))
done
echo "  $SUBMISSION_COUNT submissions for $BOOKING_COUNT bookings"

# ---------------------------------------------------------------------------
# handle_reserve: one delivery from the bookings queue.
#
# The bundle, and the whole point of the example: five things commit together,
# so there is no ordering between them left to get wrong.
# ---------------------------------------------------------------------------
handle_reserve() {
  local booking room cents txn partition lease body payload why ack_body
  booking="$(jq -r '.messages[0].data.bookingId' "$TMP/pop")"
  room="$(jq -r '.messages[0].data.room' "$TMP/pop")"
  cents="$(jq -r '.messages[0].data.cents' "$TMP/pop")"
  txn="$(jq -r '.messages[0].transactionId' "$TMP/pop")"
  partition="$(jq -r '.messages[0].partitionId' "$TMP/pop")"
  # The lease minted for THIS pop. It is what says the worker still owns the
  # message, and it is the reason the acknowledgement below can refuse.
  lease="$(jq -r '.leaseId' "$TMP/pop")"

  # `kv` and `timers` are keys of the ROOT of this body, beside `operations` and
  # not inside it. That is not a style choice: they are separate top-level
  # fields precisely so that no client can send them under one key by accident.
  #
  # kv:      required:true is what makes putIfAbsent a GATE rather than a
  #          verdict. Without it a lost race would come back applied:false while
  #          the payment and the timer went out anyway. ttlSeconds is mandatory
  #          on every KV write: a row with no expiry is a row nothing will ever
  #          delete. `forever: true` is the other legal answer, and it is
  #          exactly what an example must never write.
  # timers:  the obligation. From the moment this commits it is a row in the
  #          broker's own table, so it survives this handler, this process, this
  #          deploy and this machine. The key is ours, which is the entire
  #          reason it can be cancelled later by name. The payload is base64,
  #          and delayMs is milliseconds from now -- an absolute instant is not
  #          expressible, because deliverAt is computed in PostgreSQL and there
  #          is exactly one clock.
  # push:    partitioned by booking, so every message about one booking is in
  #          one lane.
  # ack:     carrying this delivery's lease. An expired lease refuses the ack
  #          and takes the other three down with it, which is the guarantee no
  #          compare-and-swap can give.
  payload="$(jq -rn --arg booking "$booking" --arg room "$room" \
    '{bookingId: $booking, room: $room} | tojson | @base64')"
  body="$(jq -cn --arg ns "$NS" --arg key "$(saga_key "$booking")" \
    --arg booking "$booking" --arg room "$room" --argjson cents "$cents" \
    --arg payments "$PAYMENTS" --arg expiries "$EXPIRIES" --argjson hold "$HOLD_MS" \
    --arg payload "$payload" \
    --arg txn "$txn" --arg pid "$partition" --arg grp "$RESERVER" --arg lease "$lease" '
    {operations: [{type: "push",
                   items: [{queue: $payments, partition: $booking,
                            transactionId: ("pay-" + $booking),
                            payload: {bookingId: $booking, cents: $cents}}]},
                  {type: "ack", transactionId: $txn, partitionId: $pid,
                   consumerGroup: $grp, leaseId: $lease, status: "completed"}],
     kv: [{op: "putIfAbsent", ns: $ns, key: $key,
           value: {step: "held", room: $room, cents: $cents},
           ttlSeconds: 3600, required: true}],
     timers: [{op: "schedule", queue: $expiries, timerKey: $booking,
               delayMs: $hold, txn: ("hold-" + $booking), payload: $payload}]}')"
  request POST /api/v1/transaction "$body"
  [ "$STATUS" = 200 ] || fail "the reserving bundle for $booking returned HTTP $STATUS: $(cat "$OUT")"

  # A lost gate is RETURNED, not thrown: HTTP 200 with success:false and
  # reason "kv_precondition". It is the ordinary outcome of every legitimate
  # redelivery, which makes it one of the most frequent answers this product
  # gives, and it does not belong in an error path, a retry policy or an error
  # metric -- which is exactly why it is not a 409.
  if [ "$(jq -r '.success' "$OUT")" != true ]; then
    [ "$(jq -r '.reason' "$OUT")" = kv_precondition ] \
      || fail "the reserving bundle for $booking failed: $(jq -r '.error' "$OUT")"
    # Read the verdict BEFORE the next call: $OUT is one file and the ack below
    # overwrites it. `kvReason` is the closed taxonomy of the KV refusal --
    # here `exists`, the row was already there.
    why="$(jq -r '.kvReason' "$OUT")"
    # Nothing was written: no second payment, no second timer, no second row.
    # The message still has to leave the cursor, so it is acknowledged alone.
    ack_body="$(jq -cn --arg txn "$txn" --arg pid "$partition" --arg grp "$RESERVER" --arg lease "$lease" \
      '{transactionId: $txn, partitionId: $pid, consumerGroup: $grp, leaseId: $lease, status: "completed"}')"
    request POST /api/v1/ack "$ack_body"
    [ "$STATUS" = 200 ] || fail "ack returned HTTP $STATUS"
    printf '%s %s rolled-back\n' "$RESERVER" "$booking" >> "$OBSERVED"
    echo "  $booking: already held, whole bundle rolled back ($why)"
    return 0
  fi

  printf '%s %s held\n' "$RESERVER" "$booking" >> "$OBSERVED"
  echo "  $booking: room $room held, compensation armed for $HOLD_MS ms"
}

# ---------------------------------------------------------------------------
# handle_pay: one delivery from the payments queue.
#
# A settled payment confirms the state and calls the compensation off in one
# commit; a declined card leaves the state where it is and lets the timer do its
# work.
# ---------------------------------------------------------------------------
handle_pay() {
  local booking txn partition lease state version value body timers ack_body
  booking="$(jq -r '.messages[0].data.bookingId' "$TMP/pop")"
  txn="$(jq -r '.messages[0].transactionId' "$TMP/pop")"
  partition="$(jq -r '.messages[0].partitionId' "$TMP/pop")"
  lease="$(jq -r '.leaseId' "$TMP/pop")"

  # A read in one call and a write in the next. It is safe HERE because the key
  # derives from the partition key: every message about this booking arrives in
  # one lane of this queue, and a lane has one reader per group. Where a key
  # does not derive from the partition key this shape is a race and the atomics
  # are the answer, which is exactly the compensator's situation further down.
  state="$(kv_get "$(saga_key "$booking")")"
  version="$(printf '%s' "$state" | jq -r '.version')"
  value="$(printf '%s' "$state" | jq -c '.value')"

  if [ "$booking" = "$DECLINED" ]; then
    # A declined card is a business outcome, not a delivery failure: the message
    # is done with. The room stays held, and nothing in this process is
    # responsible for giving it back.
    ack_body="$(jq -cn --arg txn "$txn" --arg pid "$partition" --arg grp "$PAYER" --arg lease "$lease" \
      '{transactionId: $txn, partitionId: $pid, consumerGroup: $grp, leaseId: $lease, status: "completed"}')"
    request POST /api/v1/ack "$ack_body"
    [ "$STATUS" = 200 ] || fail "ack returned HTTP $STATUS"
    printf '%s %s declined\n' "$PAYER" "$booking" >> "$OBSERVED"
    echo "  $booking: card declined, hold left to expire"
    return 0
  fi

  # The cancel rides the bundle: either the booking is confirmed and the
  # compensation is called off, or neither happened. Inside a transaction a
  # cancel necessarily travels in the timers array and inherits the bundle's
  # fate, which is the entire point of putting it there.
  if [ "$booking" = "$CANCEL_SKIPPED" ]; then
    timers='[]'
  else
    timers="$(jq -cn --arg expiries "$EXPIRIES" --arg booking "$booking" \
      '[{op: "cancel", queue: $expiries, timerKey: $booking}]')"
  fi

  # `expect` makes the serialisation assumption falsifiable instead of silent.
  # If the lane really serialises, it never fails and costs nothing; the day it
  # fails, two consumers are serving one partition and you learn it as a verdict
  # rather than as a wrong total.
  body="$(jq -cn --arg ns "$NS" --arg key "$(saga_key "$booking")" \
    --argjson value "$value" --argjson version "$version" --argjson timers "$timers" \
    --arg txn "$txn" --arg pid "$partition" --arg grp "$PAYER" --arg lease "$lease" '
    {operations: [{type: "ack", transactionId: $txn, partitionId: $pid,
                   consumerGroup: $grp, leaseId: $lease, status: "completed"}],
     kv: [{op: "put", ns: $ns, key: $key, value: ($value + {step: "confirmed"}),
           ttlSeconds: 3600, expect: $version, required: true}],
     timers: $timers}')"
  request POST /api/v1/transaction "$body"
  [ "$STATUS" = 200 ] || fail "the confirming bundle for $booking returned HTTP $STATUS: $(cat "$OUT")"
  [ "$(jq -r '.success' "$OUT")" = true ] \
    || fail "$booking: confirmation lost its fence ($(jq -r '.kvReason' "$OUT"))"

  printf '%s %s confirmed\n' "$PAYER" "$booking" >> "$OBSERVED"
  if [ "$booking" = "$CANCEL_SKIPPED" ]; then
    echo "  $booking: paid and confirmed, compensation deliberately NOT cancelled"
  else
    echo "  $booking: paid and confirmed, compensation cancelled"
  fi
}

# ---------------------------------------------------------------------------
# handle_compensate: one delivery from the expiries queue, which is to say one
# message a timer produced.
#
# A compensation message is not an instruction, it is a question: is this saga
# still open? A fired timer leaves no tombstone, so a cancel that arrives a
# millisecond late answers `absent` and the message goes out anyway. The state
# is the authority and it is read first.
#
# And here the key does NOT derive from the partition key: this message arrives
# on another queue entirely, in a lane that has nothing to do with the payments
# lane, so no partitioning could serialise the two writers. That is what
# `expect` is for, and on this path it is load-bearing rather than an assertion.
# ---------------------------------------------------------------------------
handle_compensate() {
  local booking room txn partition lease state step version value body ack_body
  booking="$(jq -r '.messages[0].data.bookingId' "$TMP/pop")"
  room="$(jq -r '.messages[0].data.room' "$TMP/pop")"
  txn="$(jq -r '.messages[0].transactionId' "$TMP/pop")"
  partition="$(jq -r '.messages[0].partitionId' "$TMP/pop")"
  lease="$(jq -r '.leaseId' "$TMP/pop")"

  state="$(kv_get "$(saga_key "$booking")")"
  step="$(printf '%s' "$state" | jq -r '.value.step // "gone"')"
  version="$(printf '%s' "$state" | jq -r '.version')"
  value="$(printf '%s' "$state" | jq -c '.value')"

  ack_body="$(jq -cn --arg txn "$txn" --arg pid "$partition" --arg grp "$COMPENSATOR" --arg lease "$lease" \
    '{transactionId: $txn, partitionId: $pid, consumerGroup: $grp, leaseId: $lease, status: "completed"}')"

  if [ "$step" != held ]; then
    # The booking was confirmed before this fired. Compensating here is how a
    # saga unwinds a sale that has already shipped.
    request POST /api/v1/ack "$ack_body"
    [ "$STATUS" = 200 ] || fail "ack returned HTTP $STATUS"
    printf '%s %s refused\n' "$COMPENSATOR" "$booking" >> "$OBSERVED"
    echo "  $booking: state is $step, compensation refused"
    return 0
  fi

  body="$(jq -cn --arg ns "$NS" --arg key "$(saga_key "$booking")" \
    --argjson value "$value" --argjson version "$version" \
    --arg txn "$txn" --arg pid "$partition" --arg grp "$COMPENSATOR" --arg lease "$lease" '
    {operations: [{type: "ack", transactionId: $txn, partitionId: $pid,
                   consumerGroup: $grp, leaseId: $lease, status: "completed"}],
     kv: [{op: "put", ns: $ns, key: $key, value: ($value + {step: "expired"}),
           ttlSeconds: 3600, expect: $version, required: true}]}')"
  request POST /api/v1/transaction "$body"
  [ "$STATUS" = 200 ] || fail "the compensating bundle for $booking returned HTTP $STATUS: $(cat "$OUT")"

  if [ "$(jq -r '.success' "$OUT")" != true ]; then
    # Somebody confirmed it between the read and the commit. The fence held,
    # nothing was written, and the room stays sold.
    [ "$(jq -r '.reason' "$OUT")" = kv_precondition ] \
      || fail "the compensating bundle for $booking failed: $(jq -r '.error' "$OUT")"
    request POST /api/v1/ack "$ack_body"
    [ "$STATUS" = 200 ] || fail "ack returned HTTP $STATUS"
    printf '%s %s refused\n' "$COMPENSATOR" "$booking" >> "$OBSERVED"
    echo "  $booking: confirmed under us, compensation refused by the fence"
    return 0
  fi

  printf '%s\n' "$room" >> "$RELEASED"
  printf '%s %s released\n' "$COMPENSATOR" "$booking" >> "$OBSERVED"
  echo "  $booking: hold expired, room $room released"
}

# ---------------------------------------------------------------------------
# drain <queue> <group> <handler> <deliveries> <deadline_ms>: pop and handle
# until this group has handled that many deliveries, or the deadline passes. The
# count is the bound and the deadline is the net; neither is a wait for silence.
# ---------------------------------------------------------------------------
drain() {
  local queue="$1" group="$2" handler="$3" wanted="$4" budget="$5" deadline
  deadline=$(( $(now_ms) + budget ))

  while [ "$(grep -c "^$group " "$OBSERVED" || true)" -lt "$wanted" ]; do
    [ "$(now_ms)" -lt "$deadline" ] || break

    # subscriptionMode=all is what makes a group created now read what was
    # pushed before it existed: a new cursor is seeded at the TAIL unless you
    # say otherwise. batch=1 keeps one message in flight.
    request GET "/api/v1/pop/queue/$queue?consumerGroup=$group&subscriptionMode=all&batch=1&wait=true&timeout=$POLL_MS"
    # 204 is an empty pop, with no body at all. Go round again until the
    # deadline.
    [ "$STATUS" != 204 ] || continue
    [ "$STATUS" = 200 ] || fail "pop returned HTTP $STATUS"
    cp "$OUT" "$TMP/pop"
    "$handler"
  done
}

# -------------------------------------------------------------------- reserving
echo
echo "reserving"
drain "$BOOKINGS" "$RESERVER" handle_reserve "$SUBMISSION_COUNT" "$PHASE_MS"

check "$(grep -c "^$RESERVER " "$OBSERVED" || true)" "$SUBMISSION_COUNT" \
  'the reserver reached a decision on every submission'
check "$(grep -c "^$RESERVER .* rolled-back$" "$OBSERVED" || true)" 1 \
  'the duplicate submission lost the gate exactly once'

# Pending timers are a table you can read, not a promise you have to trust.
request GET "/api/v1/timers/$EXPIRIES?limit=50"
[ "$STATUS" = 200 ] || fail "listing timers returned HTTP $STATUS"
echo "  timers armed: $(jq -r '[.rows[].timerKey] | sort | join(", ")' "$OUT")"
check "$(jq -r '.rows | length' "$OUT")" "$BOOKING_COUNT" \
  'one compensation is armed per booking and the duplicate added none'

# ----------------------------------------------------------------------- paying
echo
echo "paying"
drain "$PAYMENTS" "$PAYER" handle_pay "$BOOKING_COUNT" "$PHASE_MS"

check "$(grep -c "^$PAYER " "$OBSERVED" || true)" "$BOOKING_COUNT" \
  'every booking was asked to pay once and the duplicate produced no second payment'
check "$(awk -v g="$PAYER" '$1 == g {print $2}' "$OBSERVED" | sort -u | wc -l | tr -d ' ')" \
  "$BOOKING_COUNT" 'no booking was asked to pay twice'

# The cancel is observable before anything is delivered: the row is gone from
# the staging table. A peek is how you ask, and a miss is {"found":false} with
# HTTP 200, never a 404.
request GET "/api/v1/timers/$EXPIRIES/B-1"
[ "$STATUS" = 200 ] || fail "peek returned HTTP $STATUS"
check "$(jq -r '.found' "$OUT")" false \
  'the compensation cancelled inside the confirming bundle is gone from the table'
request GET "/api/v1/timers/$EXPIRIES/$DECLINED"
check "$(jq -r '.found' "$OUT")" true \
  "$DECLINED was never confirmed, so its compensation is still armed"
request GET "/api/v1/timers/$EXPIRIES/$CANCEL_SKIPPED"
check "$(jq -r '.found' "$OUT")" true \
  "$CANCEL_SKIPPED is confirmed but its compensation is still armed on purpose"

# ----------------------------------------------------------------- compensating
echo
echo "compensating"
# Two timers were left armed, so two messages must arrive: that is the count,
# and TIMER_DEADLINE_MS is the deadline behind it.
drain "$EXPIRIES" "$COMPENSATOR" handle_compensate 2 "$TIMER_DEADLINE_MS"
check "$(grep -c "^$COMPENSATOR " "$OBSERVED" || true)" 2 \
  'both uncancelled compensations were delivered'

# Then a bounded second pass with room for two more. It is the only honest way
# to say "a cancelled timer never arrived": the first pass would have stopped at
# two whatever those two were, so the claim is really that nothing else shows up
# afterwards.
drain "$EXPIRIES" "$COMPENSATOR" handle_compensate 4 4000

# --------------------------------------------------------------------- checking
echo
echo "checking"

check "$(grep -c "^$COMPENSATOR " "$OBSERVED" || true)" 2 \
  'nothing else arrived on a second pass: still 2 compensations'
check "$(grep -c "^$COMPENSATOR B-1 \|^$COMPENSATOR B-2 " "$OBSERVED" || true)" 0 \
  'a cancelled compensation was never delivered'
check "$(cat "$RELEASED" | tr -d ' \n')" 103 \
  'exactly one room went back on sale, the one whose card was declined'
check "$(grep -c "^$COMPENSATOR $CANCEL_SKIPPED refused$" "$OBSERVED" || true)" 1 \
  'the compensation for the confirmed booking was refused by the consumer, not prevented by the cancel'

# The saga rows are readable state, not an internal detail: a support engineer
# can answer "what happened to this booking" without a second system. getMany
# reports `missing` explicitly, because absence is a datum and not a hole
# computed by difference.
keys="$(printf '%s\n' $BOOKING_IDS | jq -R 'sub("^"; "saga:")' | jq -sc .)"
body="$(jq -cn --arg ns "$NS" --argjson keys "$keys" \
  '{operations: [{op: "getMany", ns: $ns, keys: $keys}]}')"
request POST /api/v1/kv "$body"
[ "$STATUS" = 200 ] || fail "kv getMany returned HTTP $STATUS"
check "$(jq -r '.results[0].rows | length' "$OUT")" "$BOOKING_COUNT" \
  'every booking left exactly one saga row'
check "$(jq -r '.results[0].missing | length' "$OUT")" 0 \
  'no booking is missing its saga row'
check "$(jq -r '[.results[0].rows[] | select(.value.step == "confirmed") | .key] | sort | join(",")' "$OUT")" \
  "saga:B-1,saga:B-2,saga:$CANCEL_SKIPPED" \
  "three bookings ended confirmed, $CANCEL_SKIPPED included, after its compensation was delivered"
check "$(jq -r --arg key "saga:$DECLINED" '.results[0].rows[] | select(.key == $key) | .value.step' "$OUT")" \
  expired "$DECLINED was unwound by its timer, with nobody awake to do it"

echo
echo "  final: $(jq -r '[.results[0].rows[] | (.key | sub("^saga:"; "")) + "=" + .value.step] | sort | join(", ")' "$OUT")"

echo
echo "PASS: $CHECKS checks"
```

## One bundle, and nothing left to order

The reserving handler makes one call to [`/api/v1/transaction`](/reference/http/transaction) carrying
four kinds of work: a `kv` write, a `timers` schedule, a push and an ack. `kv` and `timers` are
top-level fields of that request, beside `operations` and never inside it, and everything in the
request commits or rolls back together.

What that removes is the ordering question. Written as four calls there are permutations to pick
between and every one of them is wrong somewhere: schedule the timer first and a crash before the
push leaves a compensation for work that was never started, push first and a crash before the
schedule leaves work with no compensation at all. As one commit there is no first and no second.

The `putIfAbsent` at the front carries `required: true`, which is what makes it a gate rather than a
verdict. Without it, a duplicate submission would come back `applied: false` while the payment and
the timer went out anyway. With it, the whole bundle rolls back, acknowledgement included, and the
answer is HTTP 200 with `success: false` and `reason: "kv_precondition"`. That is **returned, not
thrown**: it is the ordinary outcome of every legitimate redelivery, so it belongs in an `if` and not
in a catch block where the reflex is to retry. The duplicate is then acknowledged on its own, because
a message that has nothing to do still has to leave the cursor.

Every KV write in the program carries `ttlSeconds`, spelled as an hour in each client's own idiom.
The expiry is mandatory: exactly one of `ttlSeconds` and `forever`, with zero being the same error as
two. None of these programs uses `forever`, which would leave rows in a shared database that nothing
removes if a run failed.

## The cancel is not the guarantee

The confirming bundle cancels the compensation timer in the same commit that writes `step:
"confirmed"` and acknowledges the payment. That is the right thing to do and it is not sufficient,
for a reason that is structural rather than a race you can shrink.

A timer's row is deleted when it fires. There is no completed-timer history, which is what makes the
fire exactly-once and leaves nothing to reconcile after a crash. The cost is that a cancel arriving
after the fire has nothing to find, and answers `absent`. The response carries the `txn` of the
message the timer was going to deliver, so the authority is available without a second API: look for
that `txn` in the destination queue. But `absent` on its own never means "not delivered".

So the compensation is written as a question rather than an instruction. The consumer reads the saga
state first, and compensates only if the saga is still `held`. The program makes that path
deterministic instead of hoping to hit it: one booking pays, is confirmed, and has its cancel
deliberately skipped, so its compensation is delivered for real and has to be refused on arrival.
It is the exact shape of the five-milliseconds-late cancel, made reproducible.

## Two fences, and only one of them is optional

Both writes after the first carry `expect` with the version the handler read. They are not the same
assertion.

In the payer, the key derives from the booking id, which is also the partition key of the payments
queue. Every message about one booking lands in one lane, and a lane has one reader per consumer
group, so the read-then-write is already serialised. `expect` there is a claim about that
serialisation, made falsifiable. If it never fails it cost nothing, and the day it fails you have
learned that two consumers are serving one partition, as a verdict rather than as a wrong total.

In the compensator the key does not derive from the partition key at all. That message arrives on a
different queue, in a lane that has nothing to do with the payments lane, so no partitioning could
serialise the two writers. There `expect` is load-bearing: it is what stops a compensation computed
from a stale read from overwriting a confirmation that landed in between. The transaction is still
the primary fence, because an expired lease refuses the acknowledgement and takes the state write
down with it, which a compare-and-swap cannot do. `expect` is the secondary one.

## What each failure does

| what happens | what the program does about it |
|---|---|
| the same booking is submitted twice | the second bundle loses the gate and rolls back whole: no second payment, no second timer, no second state row. The program asserts all three counts |
| the reserving handler dies after the commit | everything happened, including the compensation. The redelivery loses the gate and acknowledges |
| the reserving handler dies before the commit | nothing happened, including the timer. The redelivery reserves normally |
| the lease expired while the handler worked | the acknowledgement is refused and the other four operations are refused with it, so no orphan timer and no orphan hold |
| the payment settles and the cancel arrives in time | the timer's row is gone before it fires, and the compensation is never delivered. The program asserts this with a `peek` and with a second delivery pass that finds nothing |
| the payment settles and the cancel arrives too late | the compensation is delivered anyway, reads the state, finds `confirmed` and refuses. This is the case the program scripts on purpose |
| the payment is declined | the state stays `held`, nobody is awake to release the room, and the timer does it at the hold's expiry |
| a broker dies holding the timer's claim | the lease expires and another broker fires it. Worst case, delivery is late by `QUEEN_SWEEPER_LEASE_MS`. A cancel in that window answers `too_late`, which is a verdict and not a failure |

## Run it

Both surfaces this needs are served by every broker, so there is nothing to turn on and no flag to
pass. Each program probes once at the start anyway, because a `503` from [an operator's kill
switch](/deploy/state) or a `403` from a quota is worth naming before the first real call fails with
something that reads like a bug.

Against a broker from [the quickstart](/start/quickstart):

```bash
QUEEN_URL=http://localhost:6632 examples/apps/run.sh
```

Every program on this page asserts the property it exists to demonstrate and exits non-zero if it
does not hold. The runner takes a language name to run one of them on its own, for example
`examples/apps/run.sh go`. Each phase ends on a **count** with a deadline behind it rather than on a
quiet interval, because a program that stops when nothing has arrived for a while passes on a broker
that delivered nothing at all. The compensation phase gets the longest deadline of the three, since a
timer is delivered no earlier than its delay plus one sweep. And each run purges what it wrote:
the queues, the saga rows, and any timer still armed, because a pending timer is keyed by name in a
table of its own and deleting the queue does not reach it.

Source: https://queenmq.com/use/full-examples/saga/index.mdx