Pop a batch or run a consume loop: queue mode and consumer groups, where a new group starts, batches, leases and renewal, conflation, seek, and reads by offset.

There are two ways to read. consume() is a loop: it takes messages, calls your handler, acks them when the handler returns, and goes on. pop() is one call: it hands you the messages and leaves the ack to you.

Why

most workers are loops, and consume() writes the loop for you, with the error handling done right. pop() is for the cases a loop does not fit: a scheduled job that reads once, a script, or code that wants to decide itself when to ack.

The consume loop. It pops a batch, which leases the partition; calls your handler; acks when the handler returns, or fails the message when it throws; and goes back to pop. With autoAck off, your handler acks itself, often inside a transaction. pop() is one turn of the loop, done by hand. — What consume() does for you.

Queue mode and consumer groups

Without a consumer group, each message goes to one worker, and the broker sizes every pop for you.

await queen.queue('orders')
  .each()                         // the handler gets one message at a time
  .limit(1)                       // stop after one message; a worker leaves this out
  .consume(async (msg) => {       // acked when the handler returns, retried if it throws
    console.log(msg.data)
  })

async def handle(msg):                 # acked when it returns, retried if it raises
    print(msg["data"])

await (
    queen.queue("orders")
    .each()                            # the handler gets one message at a time
    .limit(1)                          # stop after one message; a worker leaves this out
    .consume(handle)
)

err = client.Queue("orders").
	Each().   // the handler gets one message at a time
	Limit(1). // stop after one message; a worker leaves this out
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		fmt.Println(msg.Data) // acked when this returns nil, retried on an error
		return nil
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}

use queen_mq::Message;

queen
    .queue("orders")
    .limit(1) // stop after one message; a worker leaves this out
    .consume(|msg: Message| async move {
        println!("{}", msg.data); // acked when this returns Ok, retried on an Err
        Ok::<_, queen_mq::Error>(())
    })
    .await?;

$queen->queue('orders')
    ->each()                          // the handler gets one message at a time
    ->limit(1)                        // stop after one message; a worker leaves this out
    ->consume(function (array $msg) { // acked when it returns, retried if it throws
        print_r($msg['data']);
    })
    ->execute();

client.queue("orders")
    .each()                        // the handler gets one message at a time
    .limit(1)                      // stop after one message; a worker leaves this out
    .consume([](const json& msg) { // acked when it returns, retried if it throws
        std::cout << msg["data"] << std::endl;
    });

With a consumer group, every group gets every message and the workers of one group share them. A new group starts at the tail, so this one asks for everything already in the queue. Here every option is set by hand, and the handler acks.

await queen.queue('orders')
  .group('billing')               // every group gets every message
  .subscriptionMode('all')        // a new group starts at the tail unless told otherwise
  .batch(100)                     // up to 100 messages per pop...
  .partitions(10)                 // ...from up to 10 partitions
  .renewLease(true, 2000)         // heartbeat every 2 s while the handler runs
  .autoAck(false)                 // the handler decides the outcome
  .each()
  .limit(1)
  .consume(async (msg) => {
    const ok = msg.data.total > 0
    await queen.ack(msg, ok ? 'completed' : 'failed', { group: 'billing' })
  })

async def bill(msg):
    ok = msg["data"]["total"] > 0
    await queen.ack(msg, "completed" if ok else "failed", {"group": "billing"})

await (
    queen.queue("orders")
    .group("billing")                  # every group gets every message
    .subscription_mode("all")          # a new group starts at the tail unless told otherwise
    .batch(100)                        # up to 100 messages per pop...
    .partitions(10)                    # ...from up to 10 partitions
    .renew_lease(True, 2000)           # heartbeat every 2 s while the handler runs
    .auto_ack(False)                   # the handler decides the outcome
    .each()
    .limit(1)
    .consume(bill)
)

err = client.Queue("orders").
	Group("billing").                            // every group gets every message
	SubscriptionMode(queen.SubscriptionModeAll). // a new group starts at the tail unless told otherwise
	Batch(100).                                  // up to 100 messages per pop...
	Partitions(10).                              // ...from up to 10 partitions
	RenewLease(true, 2000).                      // heartbeat every 2 s while the handler runs
	AutoAck(false).                              // the handler decides the outcome
	Each().
	Limit(1).
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		ok := msg.Data["total"].(float64) > 0
		_, err := client.Ack(ctx, msg, ok, queen.AckOptions{ConsumerGroup: "billing"})
		return err
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}

use queen_mq::{AckStatus, SubscriptionMode};

let client = queen.clone();
queen
    .queue("orders")
    .group("billing")                     // every group gets every message
    .subscription_mode(SubscriptionMode::All) // a new group starts at the tail unless told otherwise
    .batch(100)                           // up to 100 messages per pop...
    .partitions(10)                       // ...from up to 10 partitions
    .renew_lease(Duration::from_secs(2))  // heartbeat every 2 s while the handler runs
    .auto_ack(false)                      // the handler decides the outcome
    .limit(1)
    .consume(move |msg: Message| {
        let client = client.clone();
        async move {
            let ok = msg.data["total"].as_f64().unwrap_or(0.0) > 0.0;
            let status = if ok { AckStatus::Completed } else { AckStatus::Failed };
            client.ack_with(&msg, status, None).await?;
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;

$queen->queue('orders')
    ->group('billing')                // every group gets every message
    ->subscriptionMode('all')         // a new group starts at the tail unless told otherwise
    ->batch(100)                      // up to 100 messages per pop...
    ->partitions(10)                  // ...from up to 10 partitions
    ->renewLease(true, 2000)          // renew every 2 s, between messages
    ->autoAck(false)                  // the handler decides the outcome
    ->each()
    ->limit(1)
    ->consume(function (array $msg) use ($queen) {
        $ok = $msg['data']['total'] > 0;
        $queen->ack($msg, $ok ? 'completed' : 'failed', ['group' => 'billing']);
    })
    ->execute();

client.queue("orders")
    .group("billing")          // every group gets every message
    .subscription_mode("all")  // a new group starts at the tail unless told otherwise
    .batch(100)                // up to 100 messages per pop...
    .partitions(10)            // ...from up to 10 partitions
    .renew_lease(true, 2000)   // heartbeat every 2 s while the handler runs
    .auto_ack(false)           // the handler decides the outcome
    .each()
    .limit(1)
    .consume([&](const json& msg) {
        bool ok = msg["data"]["total"].get<double>() > 0;
        client.ack(msg, ok, {{"group", "billing"}});  // true: completed, false: failed
    });

Ack failed and the message comes back, up to the queue’s retryLimit; after the last retry it goes to the dead-letter queue.

Why

one queue often feeds several services: billing, emails and analytics all want every order. Each service is a consumer group, reading at its own pace, and none of them can slow the others down. Inside a group, the workers compete, so you add workers to go faster.

Left, queue mode with no consumer group: the partitions of orders are shared out, and each message goes to exactly one of the workers w1, w2 and w3. Right, three consumer groups, billing, emails and analytics, on the same queue: each group gets every message and keeps its own position; inside billing, two workers split the partitions between them. — Queue mode shares the messages; every consumer group gets all of them.

One message at a time, or a whole batch

With .each(), the handler gets one message at a time. Without it, the handler gets the whole batch as an array.

await queen.queue('events')
  .group('archiver')
  .subscriptionMode('all')
  .batch(100)                         // up to 100 messages per call
  .limit(1000)
  .consume(async (messages) => {      // no .each(): the handler gets the whole batch
    console.log('archiving', messages.length)
  })                                  // all acked if it returns, all nacked if it throws

async def archive(messages):          # no .each(): the handler gets the whole batch
    print("archiving", len(messages))

await (
    queen.queue("events")
    .group("archiver")
    .subscription_mode("all")
    .batch(100)                       # up to 100 messages per call
    .limit(1000)
    .consume(archive)                 # all acked if it returns, all nacked if it raises
)

err = client.Queue("events").
	Group("archiver").
	SubscriptionMode(queen.SubscriptionModeAll).
	Batch(100). // up to 100 messages per call
	Limit(1000).
	// ConsumeBatch, not Consume: the handler gets the whole batch
	ConsumeBatch(ctx, func(ctx context.Context, messages []*queen.Message) error {
		fmt.Println("archiving", len(messages))
		return nil // all acked if it returns nil, all nacked if it returns an error
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}

queen
    .queue("events")
    .group("archiver")
    .subscription_mode(SubscriptionMode::All)
    .batch(100)                             // up to 100 messages per call
    .limit(1000)
    .consume_batch(|messages: Vec<Message>| async move {
        println!("archiving {}", messages.len()); // the handler gets the whole batch
        Ok::<_, queen_mq::Error>(())
    })                                      // all acked on Ok, all nacked on an Err
    .await?;

$queen->queue('events')
    ->group('archiver')
    ->subscriptionMode('all')
    ->batch(100)                         // up to 100 messages per call
    ->limit(1000)
    ->consume(function (array $messages) { // no ->each(): the handler gets the whole batch
        echo 'archiving ', count($messages), "\n";
    })                                   // all acked if it returns, all nacked if it throws
    ->execute();

client.queue("events")
    .group("archiver")
    .subscription_mode("all")
    .batch(100)                          // up to 100 messages per call
    .limit(1000)
    .consume([&](const json& messages) { // no each(): the handler gets the whole batch
        std::cout << "archiving " << messages.size() << "\n";
    });                                  // all acked if it returns, nacked if it throws

What the client does with your handler’s result (autoAck, on by default):

It returns: the message (or the whole batch) is acked.
It throws: the message (or the whole batch) is nacked as failed, which spends one retry, and the loop goes on with the next messages.
It throws under .each(): the rest of that partition’s messages in the same batch are skipped, because they will be delivered again after the failed one, in order. Messages of other partitions in the batch are handled as usual.
With .autoAck(false): the handler acks itself (in a transaction, for example). A handler that throws is still nacked, so its partition is not stuck until the lease runs out.

In the other SDKs: the list above is what JavaScript, Python and Go do. Under autoAck(false), Rust and C++ do not nack a handler that fails (its messages wait for their lease), and PHP stops the loop: catch the error and ack failed yourself.

Why

a batch is cheaper (one ack for a hundred messages, one database write for a hundred rows), and one at a time is simpler and fails one message instead of a hundred. Pick per worker.

If you add .onSuccess() or .onError() to a consumer, the client stops acking and nacking for you: you decide in those callbacks what happens to each message. Python and PHP have the same callbacks (PHP with autoAck(false)); Go, Rust and C++ have none, so their code turns autoAck off and acks in the handler.

await queen.queue('emails')
  .group('mailer')
  .subscriptionMode('all')
  .each()
  .limit(1)
  .consume(async (msg) => sendEmail(msg.data))
  .onSuccess((msg) => queen.ack(msg, 'completed', { group: 'mailer' }))
  .onError((msg, error) => queen.ack(msg, error.permanent ? 'dlq' : 'failed',
    { group: 'mailer', error: error.message }))

async def deliver(msg):
    await send_email(msg["data"])

async def completed(msg, result):
    await queen.ack(msg, "completed", {"group": "mailer"})

async def failed(msg, error):
    status = "dlq" if getattr(error, "permanent", False) else "failed"
    await queen.ack(msg, status, {"group": "mailer", "error": str(error)})

await (
    queen.queue("emails")
    .group("mailer")
    .subscription_mode("all")
    .each()
    .limit(1)
    .consume(deliver)
    .on_success(completed)
    .on_error(failed)
)

err = client.Queue("emails").
	Group("mailer").
	SubscriptionMode(queen.SubscriptionModeAll).
	AutoAck(false). // the client stops acking: the handler gives every answer
	Each().
	Limit(1).
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		opts := queen.AckOptions{ConsumerGroup: "mailer"}
		err := sendEmail(msg.Data)
		switch {
		case err == nil:
			_, err = client.Ack(ctx, msg, true, opts)
		case errors.Is(err, errPermanent): // trying again cannot help
			// Ack takes no dlq status: a transaction carries it, without the error text.
			_, err = client.Transaction().Ack(msg, "dlq", opts).Commit(ctx)
		default:
			opts.Error = err.Error()
			_, err = client.Ack(ctx, msg, false, opts) // failed: back after retryDelay
		}
		return err
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}

let client = queen.clone();
queen
    .queue("emails")
    .group("mailer")
    .subscription_mode(SubscriptionMode::All)
    .limit(1)
    .auto_ack(false) // no callbacks in the Rust client: the handler settles each message
    .consume(move |msg: Message| {
        let client = client.clone();
        async move {
            let (status, error) = match send_email(&msg.data).await {
                Ok(()) => (AckStatus::Completed, None),
                Err(e) if e.permanent => (AckStatus::Dlq, Some(e.message)),
                Err(e) => (AckStatus::Failed, Some(e.message)),
            };
            client.ack_with(&msg, status, error).await?;
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;

$queen->queue('emails')
    ->group('mailer')
    ->subscriptionMode('all')
    ->autoAck(false)                     // PHP: the callbacks below ack, not the loop
    ->each()
    ->limit(1)
    ->consume(fn (array $msg) => sendEmail($msg['data']))
    ->onSuccess(fn (array $msg) => $queen->ack($msg, 'completed', ['group' => 'mailer']))
    ->onError(fn (array $msg, Throwable $error) => $queen->ack($msg,
        $error instanceof PermanentError ? 'dlq' : 'failed',
        ['group' => 'mailer', 'error' => $error->getMessage()]))
    ->execute();                         // with autoAck(false), a throw also ends execute()

client.queue("emails")
    .group("mailer")
    .subscription_mode("all")
    .auto_ack(false)                        // the handler settles each message itself
    .each()
    .limit(1)
    .consume([&](const json& msg) {
        try {
            send_email(msg["data"]);
            client.ack(msg, true, {{"group", "mailer"}});
        } catch (const PermanentError& e) { // trying again cannot help
            // QueenClient::ack sends completed or failed only, so the dlq answer
            // goes to the broker's ack route through the client's HTTP layer.
            client.get_http_client()->post("/api/v1/ack", {
                {"transactionId", msg["transactionId"]}, {"leaseId", msg["leaseId"]},
                {"partitionId", msg["partitionId"]}, {"consumerGroup", "mailer"},
                {"status", "dlq"}, {"error", e.what()}});
        } catch (const std::exception& e) { // may work later: spend a retry
            client.ack(msg, false, {{"group", "mailer"}, {"error", e.what()}});
        }
    });

Why

some failures should be retried and some never will succeed (an email address that does not exist). The callbacks let you send the first kind back with failed and the second kind straight to the dead-letter queue with dlq.

How many messages a pop takes: autopilot

By default the broker decides how many messages each pop gets (batch, from 100 to 1,000, based on how fast the group acks) and from how many partitions (partitions, the ones that have work, shared among the waiting workers). A value you set yourself is never changed. partitions(N) lets one pop take messages from up to N partitions (1 to 64), with batch as the total. .autopilot(false) goes back to one message from one partition per pop.

Why

the right batch size depends on your handler and your load, and it changes during the day. The broker sees both, so it can size every pop better than a number fixed in the code.

Leases and long work

A pop leases its messages to the worker for the queue’s leaseTime. Work that can take longer renews the lease while it runs: .renewLease(true, 10000) renews every 10 seconds, and queen.renew(message) renews once. In PHP, the renewal happens between messages, so a single long handler calls the renew callback it is given. A lease that has already expired cannot be renewed: its messages go to the next worker, marked as one delivery attempt more.

Why

the lease is what protects order and stops two workers from doing the same work. A short lease gets a crashed worker’s messages moving again quickly; renewing lets a slow but healthy worker keep them.

Where a new group starts

A group’s starting point is decided on its first pop and then stored: new (the default) reads only what is pushed after that first pop, all reads everything still kept, and subscriptionFrom reads from a point in time. A group that already exists moves with a seek, to a time or to the end.

await queen.admin.seekConsumerGroup('archiver', 'events', {
  timestamp: '2026-10-22T00:00:00Z',
})
// the archiver group now reads again everything pushed since that time

await queen.admin.seekConsumerGroup('archiver', 'events', { toEnd: true })
// and now it skips everything it has not read yet

await queen.admin.seek_consumer_group("archiver", "events", {
    "timestamp": "2026-10-22T00:00:00Z",
})
# the archiver group now reads again everything pushed since that time

await queen.admin.seek_consumer_group("archiver", "events", {"toEnd": True})
# and now it skips everything it has not read yet

_, err = client.Admin().SeekConsumerGroup(ctx, "archiver", "events",
	queen.SeekConsumerGroupOptions{Timestamp: "2026-10-22T00:00:00Z"})
if err != nil {
	log.Fatal(err)
}
// the archiver group now reads again everything pushed since that time

_, err = client.Admin().SeekConsumerGroup(ctx, "archiver", "events",
	queen.SeekConsumerGroupOptions{ToEnd: true})
if err != nil {
	log.Fatal(err)
}
// and now it skips everything it has not read yet

use queen_mq::SeekRequest;

let since = SeekRequest {
    timestamp: Some("2026-10-22T00:00:00Z".into()),
    position: None,
};
queen.admin().seek_consumer_group("archiver", "events", &since).await?;
// the archiver group now reads again everything pushed since that time

// SeekRequest has no toEnd in the Rust client: a time after the newest
// message moves the cursor to the same place.
let to_end = SeekRequest {
    timestamp: Some("9999-12-31T00:00:00Z".into()),
    position: None,
};
queen.admin().seek_consumer_group("archiver", "events", &to_end).await?;
// and now it skips everything it has not read yet

$queen->admin()->seekConsumerGroup('archiver', 'events', [
    'timestamp' => '2026-10-22T00:00:00Z',
]);
// the archiver group now reads again everything pushed since that time

$queen->admin()->seekConsumerGroup('archiver', 'events', ['toEnd' => true]);
// and now it skips everything it has not read yet

// The C++ client has no admin calls: a seek is a POST to the broker's
// consumer-group route, through the client's own HTTP layer.
auto http = client.get_http_client();
http->post("/api/v1/consumer-groups/archiver/queues/events/seek",
           {{"timestamp", "2026-10-22T00:00:00Z"}});
// the archiver group now reads again everything pushed since that time

http->post("/api/v1/consumer-groups/archiver/queues/events/seek", {{"toEnd", true}});
// and now it skips everything it has not read yet

Why

a new service usually wants only new events, a new projection wants the whole history, and a bug fix wants to replay the last day. A seek to the end is how a group skips a backlog it no longer needs.

Only the newest message: conflation

A group with .conflation() gets only the newest message of each partition, and its cursor moves past the older ones.

const [latest] = await queen.queue('prices')
  .group('repricer')
  .subscriptionMode('all')
  .conflation()                       // only the newest message of each partition
  .wait(false)
  .pop()
console.log(latest.data)              // { sku: 'A-1', price: 12 }: 10 and 11 are skipped
await queen.ack(latest, 'completed', { group: 'repricer' })

[latest] = await (
    queen.queue("prices")
    .group("repricer")
    .subscription_mode("all")
    .conflation()                     # only the newest message of each partition
    .wait(False)
    .pop()
)
print(latest["data"])                 # {'sku': 'A-1', 'price': 12}: 10 and 11 are skipped
await queen.ack(latest, "completed", {"group": "repricer"})

newest, err := client.Queue("prices").
	Group("repricer").
	SubscriptionMode(queen.SubscriptionModeAll).
	Conflation(true). // only the newest message of each partition
	Wait(false).
	Pop(ctx)
if err != nil {
	log.Fatal(err)
}
latest := newest[0]
fmt.Println(latest.Data) // map[price:12 sku:A-1]: 10 and 11 are skipped
_, err = client.Ack(ctx, latest, true, queen.AckOptions{ConsumerGroup: "repricer"})
if err != nil {
	log.Fatal(err)
}

let newest = queen
    .queue("prices")
    .group("repricer")
    .subscription_mode(SubscriptionMode::All)
    .conflation(true)                       // only the newest message of each partition
    .wait(false)
    .pop()
    .await?;
let latest = &newest[0];
println!("{}", latest.data); // {"price":12,"sku":"A-1"}: 10 and 11 are skipped
queen.ack(latest).await?;

[$latest] = $queen->queue('prices')
    ->group('repricer')
    ->subscriptionMode('all')
    ->conflation()                       // only the newest message of each partition
    ->wait(false)
    ->pop();
print_r($latest['data']);                // [sku => A-1, price => 12]: 10 and 11 are skipped
$queen->ack($latest, 'completed', ['group' => 'repricer']);

json latest = client.queue("prices")
    .group("repricer")
    .subscription_mode("all")
    .conflation()                    // only the newest message of each partition
    .wait(false)
    .pop()
    .at(0);
std::cout << latest["data"] << "\n"; // {"price":12,"sku":"A-1"}: 10 and 11 are skipped
client.ack(latest, true, {{"group", "repricer"}});

Why

some queues carry commands where only the latest one matters: “recompute the price of A-1”. A worker that is behind should do the work once, with the newest input, instead of three times with inputs that are already old.

Conflation is decided by the group’s first pop, needs a consumer group, and cannot be combined with commitOnDelivery.

Faster, with no ack: commitOnDelivery

A pop with .commitOnDelivery() moves the group’s cursor as the messages are handed out. There is no lease and nothing to ack.

const readings = await queen.queue('metrics')
  .group('live-chart')
  .subscriptionMode('all')
  .commitOnDelivery()                 // the cursor moves as they are handed out: no ack
  .batch(100)
  .wait(false)
  .pop()

readings = await (
    queen.queue("metrics")
    .group("live-chart")
    .subscription_mode("all")
    .commit_on_delivery()             # the cursor moves as they are handed out: no ack
    .batch(100)
    .wait(False)
    .pop()
)

readings, err := client.Queue("metrics").
	Group("live-chart").
	SubscriptionMode(queen.SubscriptionModeAll).
	CommitOnDelivery(true). // the cursor moves as they are handed out: no ack
	Batch(100).
	Wait(false).
	Pop(ctx)
if err != nil {
	log.Fatal(err)
}

let readings = queen
    .queue("metrics")
    .group("live-chart")
    .subscription_mode(SubscriptionMode::All)
    .commit_on_delivery(true) // the cursor moves as they are handed out: no ack
    .batch(100)
    .wait(false)
    .pop()
    .await?;

$readings = $queen->queue('metrics')
    ->group('live-chart')
    ->subscriptionMode('all')
    ->commitOnDelivery()                 // the cursor moves as they are handed out: no ack
    ->batch(100)
    ->wait(false)
    ->pop();

json readings = client.queue("metrics")
    .group("live-chart")
    .subscription_mode("all")
    .commit_on_delivery()  // the cursor moves as they are handed out: no ack
    .batch(100)
    .wait(false)
    .pop();

Why

a live dashboard or a metrics chart can lose a reading now and then, and would rather not pay for leases and acks. This is at-most-once delivery: a worker that crashes after the pop loses that batch. It works with pop() only; consume() refuses it.

Running a consumer for real

const stop = new AbortController()
process.once('SIGTERM', () => stop.abort())     // a deploy stops the worker cleanly

await queen.queue('tasks')
  .group('workers')
  .subscriptionMode('all')
  .concurrency(4)                               // four handlers at once, on four partitions
  .renewLease(true, 10000)                      // long tasks keep their lease alive
  .idleMillis(60000)                            // or stop after a minute with nothing to do
  .each()
  .consume(async (task) => runTask(task.data), { signal: stop.signal })

stop = asyncio.Event()
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGTERM, stop.set)   # a deploy stops the worker cleanly

async def handle(task):
    await run_task(task["data"])

await (
    queen.queue("tasks")
    .group("workers")
    .subscription_mode("all")
    .concurrency(4)                   # four handlers at once, on four partitions
    .renew_lease(True, 10000)         # long tasks keep their lease alive
    .idle_millis(60000)               # or stop after a minute with nothing to do
    .each()
    .consume(handle, signal=stop)
)

ctx, stop := signal.NotifyContext(ctx, syscall.SIGTERM) // a deploy stops it cleanly
defer stop()

err := client.Queue("tasks").
	Group("workers").
	SubscriptionMode(queen.SubscriptionModeAll).
	Concurrency(4).          // four handlers at once, on four partitions
	RenewLease(true, 10000). // long tasks keep their lease alive
	IdleMillis(60000).       // or stop after a minute with nothing to do
	Each().
	Consume(ctx, func(ctx context.Context, task *queen.Message) error {
		return runTask(task.Data)
	}).
	Execute(ctx) // returns once ctx is cancelled
if err != nil {
	log.Fatal(err)
}

use queen_mq::Cancel;
use tokio::signal::unix::{signal, SignalKind};

let stop = Cancel::new();
let on_sigterm = stop.clone();
tokio::spawn(async move {                   // a deploy stops the worker cleanly
    if let Ok(mut sigterm) = signal(SignalKind::terminate()) {
        sigterm.recv().await;
        on_sigterm.cancel();
    }
});

queen
    .queue("tasks")
    .group("workers")
    .subscription_mode(SubscriptionMode::All)
    .concurrency(4)                         // four handlers at once, on four partitions
    .renew_lease(Duration::from_secs(10))   // long tasks keep their lease alive
    .idle(Duration::from_secs(60))          // or stop after a minute with nothing to do
    .poll_timeout(Duration::from_secs(5))   // a stop waits for the poll in flight (5 s)
    .cancel(stop)
    .consume(|task: Message| async move { run_task(&task.data).await })
    .await?;

// A deploy's SIGTERM stops the loop cleanly: consume() handles it itself (ext-pcntl).
$queen->queue('tasks')
    ->group('workers')
    ->subscriptionMode('all')
    ->concurrency(4)                     // four pops at once; their handlers run in turn
    ->renewLease(true, 10000)            // leases renewed every 10 s, between tasks
    ->idleMillis(60000)                  // or stop after a minute with nothing to do
    ->each()
    ->consume(fn (array $task) => runTask($task['data']))
    ->execute();

static std::atomic<bool> stop{false};            // static: a signal handler sees it
std::signal(SIGTERM, [](int) { stop = true; });  // a deploy stops the worker cleanly

client.queue("tasks")
    .group("workers")
    .subscription_mode("all")
    .concurrency(4)              // four handlers at once, on four partitions
    .renew_lease(true, 10000)    // long tasks keep their lease alive
    .idle_millis(60000)          // or stop after a minute with nothing to do
    .wait(false)                 // short polls: stop and idle are read between them
    .each()
    .consume([](const json& task) { run_task(task["data"]); }, &stop);

concurrency(n) runs n handlers at once, each on its own partitions.
limit(n) stops after n messages, and idleMillis(ms) stops after that long with nothing to do.
wait(true) (the default) waits for messages instead of returning empty, up to timeoutMillis(ms) (30 seconds by default, at most 60).
A signal stops the loop at once. Messages it had taken but not started go back with a retry ack, which spends no retry, so a deploy does not leave partitions waiting for their leases to run out.

In the other SDKs: in 2.6.0 only JavaScript gives the unstarted messages back. Python, Go, Rust, PHP and C++ stop after the message they are handling, and the rest wait for their lease. Python, Rust and C++ also notice the stop only between two long polls, so their code sets a short poll timeout. PHP stops on SIGTERM by itself.

Why

a worker runs for weeks, through deploys and quiet nights. These options decide how much it does at once, when it stops, and how it gives its work back when it is told to.

Read it again

Consuming does not delete: the queue is a log, kept for its retention. A new group can start from a point in time and read the same orders again, while the groups that already read them stay where they are.

await queen.queue('orders')
  .group('audit')                           // a new group, with its own cursor
  .subscriptionFrom('2026-10-22T00:00:00Z') // everything pushed from this time on
  .each()
  .limit(1)
  .consume(async (msg) => {
    console.log('again:', msg.data)
  })

async def audit(msg):
    print("again:", msg["data"])

await (
    queen.queue("orders")
    .group("audit")                              # a new group, with its own cursor
    .subscription_from("2026-10-22T00:00:00Z")   # everything pushed from this time on
    .each()
    .limit(1)
    .consume(audit)
)

err = client.Queue("orders").
	Group("audit").                           // a new group, with its own cursor
	SubscriptionFrom("2026-10-22T00:00:00Z"). // everything pushed from this time on
	Each().
	Limit(1).
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		fmt.Println("again:", msg.Data)
		return nil
	}).
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}

queen
    .queue("orders")
    .group("audit")                            // a new group, with its own cursor
    .subscription_from("2026-10-22T00:00:00Z") // everything pushed from this time on
    .limit(1)
    .consume(move |msg: Message| {
        async move {
            println!("again: {}", msg.data);
            Ok::<_, queen_mq::Error>(())
        }
    })
    .await?;

$queen->queue('orders')
    ->group('audit')                           // a new group, with its own cursor
    ->subscriptionFrom('2026-10-22T00:00:00Z') // everything pushed from this time on
    ->each()
    ->limit(1)
    ->consume(function (array $msg) {
        print_r($msg['data']);
    })
    ->execute();

client.queue("orders")
    .group("audit")                              // a new group, with its own cursor
    .subscription_from("2026-10-22T00:00:00Z")   // everything pushed from this time on
    .each()
    .limit(1)
    .consume([&](const json& msg) {
        std::cout << "again: " << msg["data"] << std::endl;
    });

Or read a partition by offset with fetch: no group, no lease, no ack. It suits an audit, rebuilding one entity, or a sink that keeps its own position.

const read = await queen.queue('orders').fetch(['customer-42']) // no group, lease or ack
for (const record of read.get('customer-42').records) {
  console.log(record.offset, record.payload)
}

read = await queen.queue("orders").fetch(["customer-42"])  # no group, lease or ack
for record in read["customer-42"]["records"]:
    print(record["offset"], record["payload"])

read, err := client.Queue("orders").Fetch(ctx, []string{"customer-42"}) // no group, lease or ack
if err != nil {
	log.Fatal(err)
}
for _, record := range read["customer-42"].Records {
	fmt.Println(record.Offset, string(record.Payload))
}

let read = queen.queue("orders").fetch(["customer-42"]).await?; // no group, lease or ack
for record in &read["customer-42"].records {
    println!("{} {}", record.offset, record.payload);
}

$read = $queen->queue('orders')->fetch(['customer-42']); // no group, lease or ack
foreach ($read['customer-42']['records'] as $record) {
    echo $record['offset'], ' ', json_encode($record['payload']), "\n";
}

auto read = client.queue("orders").fetch({"customer-42"});  // no group, lease or ack
for (const queen::FetchedRecord& record : read["customer-42"].records) {
    std::cout << record.offset << " " << record.payload << std::endl;
}

Why

history is useful long after the first reader is done with it: to build a new view, to audit what happened, or to rebuild one customer’s state from its events.

Pop and consume messages with consumer groups