Partitions are for ordering. Not for scaling.

1. 1

   Run a node

   One binary serves the HTTP API and the dashboard on one port. Open localhost:6632 and watch every step below land.

   $ docker run -p 6632:6632 ghcr.io/queen-mq/queen:2.6.0

   Then add the client:

   $ npm install queen-mq

   $ pip install queen-mq

   $ go get github.com/smartpricing/queen/clients/client-go/v2

   $ cargo add queen-mq serde_json
   $ cargo add tokio --features rt-multi-thread,macros

   $ composer require queen-mq/php-client

   $ # header-only: clients/client-cpp/queen_client.hpp, with cpp-httplib and nlohmann/json

   

2. 2

   Configure a queue

   Optional: the first push creates a queue with the defaults. Configure it when you want other settings.

   ConfigureJavaScript

   import { Queen } from 'queen-mq'
   
   const queen = new Queen({ url: 'http://localhost:6632' })
   
   await queen.queue('orders')
     .namespace('shop')              // groups queues in the dashboard
     .config({
       leaseTime: 30,                // seconds a worker holds a message before it is redelivered
       retryLimit: 3,                // failures before the message goes to the dead-letter queue
       retentionEnabled: true,       // retention runs only when this is on
       retentionSeconds: 86400,      // then messages older than a day are removed
     })
     .create()

   ConfigurePython

   from queen import Queen
   
   queen = Queen(url="http://localhost:6632")
   
   await (
       queen.queue("orders")
       .namespace("shop")                 # groups queues in the dashboard
       .config({
           "leaseTime": 30,               # seconds a worker holds a message before it is redelivered
           "retryLimit": 3,               # failures before the message goes to the dead-letter queue
           "retentionEnabled": True,      # retention runs only when this is on
           "retentionSeconds": 86400,     # then messages older than a day are removed
       })
       .create()
   )

   ConfigureGo

   import queen "github.com/smartpricing/queen/clients/client-go/v2"
   
   ctx := context.Background()
   client, err := queen.New("http://localhost:6632")
   if err != nil {
   	log.Fatal(err)
   }
   
   _, err = client.Queue("orders").
   	Namespace("shop"). // groups queues in the dashboard
   	Config(queen.QueueConfig{
   		LeaseTime:        30,    // seconds a worker holds a message before it is redelivered
   		RetryLimit:       3,     // failures before the message goes to the dead-letter queue
   		RetentionEnabled: true,  // retention runs only when this is on
   		RetentionSeconds: 86400, // then messages older than a day are removed
   	}).
   	Create().
   	Execute(ctx)
   if err != nil {
   	log.Fatal(err)
   }

   ConfigureRust

   use queen_mq::{Config, Queen, QueueOptions};
   
   let queen = Queen::connect(Config::new("http://localhost:6632"))?;
   
   queen
       .queue("orders")
       .namespace("shop") // groups queues in the dashboard
       .configure(QueueOptions {
           lease_time: Some(30),           // seconds a worker holds a message before it is redelivered
           retry_limit: Some(3),           // failures before the message goes to the dead-letter queue
           retention_enabled: Some(true),  // retention runs only when this is on
           retention_seconds: Some(86400), // then messages older than a day are removed
           ..Default::default()
       })
       .await?;

   ConfigurePHP

   use Queen\Queen;
   
   $queen = new Queen('http://localhost:6632');
   
   $queen->queue('orders')
       ->namespace('shop')              // groups queues in the dashboard
       ->config([
           'leaseTime' => 30,           // seconds a worker holds a message before it is redelivered
           'retryLimit' => 3,           // failures before the message goes to the dead-letter queue
           'retentionEnabled' => true,  // retention runs only when this is on
           'retentionSeconds' => 86400, // then messages older than a day are removed
       ])
       ->create()
       ->execute();

   ConfigureC++

   #include "queen_client.hpp"
   
   using json = nlohmann::json;
   
   queen::QueenClient client("http://localhost:6632");
   
   // QueueConfig has no retentionEnabled, so the options go to the broker's
   // configure route through the client's own HTTP layer.
   client.get_http_client()->post("/api/v1/configure", {
       {"queue", "orders"},
       {"namespace", "shop"},              // groups queues in the dashboard
       {"options", {
           {"leaseTime", 30},              // seconds a worker holds a message,
                                           // then it is redelivered
           {"retryLimit", 3},              // failures before the message goes
                                           // to the dead-letter queue
           {"retentionEnabled", true},     // retention runs only when this is on
           {"retentionSeconds", 86400},    // then messages older than a day are removed
       }},
   });

   Every queue option →

3. 3

   Push to a partition

   In Kafka you choose a number of partitions and hash keys onto them, so a slow customer holds up every customer that shares its partition. Here a partition is one entity: a customer, an order, a conversation. The push that names it creates it, order holds inside it, and one queue has held ten million.

   The transactionId makes a retry safe: the same id within the dedup window (an hour by default) is not written twice.

   PushJavaScript

   await queen.queue('orders')
     .partition('customer-42')       // one ordered partition per customer, created by this push
     .push([{
       transactionId: 'order-1001',  // the dedup key: pushing the same id again is a no-op
       data: { orderId: 1001, customer: 42, total: 59.9 },
     }])

   PushPython

   await (
       queen.queue("orders")
       .partition("customer-42")          # one ordered partition per customer, created by this push
       .push([{
           "transactionId": "order-1001",  # the dedup key: pushing the same id again is a no-op
           "data": {"orderId": 1001, "customer": 42, "total": 59.9},
       }])
   )

   PushGo

   _, err = client.Queue("orders").
   	Partition("customer-42"). // one ordered partition per customer, created by this push
   	Push(map[string]interface{}{"orderId": 1001, "customer": 42, "total": 59.9}).
   	TransactionID("order-1001"). // the dedup key: pushing the same id again is a no-op
   	Execute(ctx)
   if err != nil {
   	log.Fatal(err)
   }

   PushRust

   use queen_mq::PushItem;
   
   queen
       .queue("orders")
       .push_items(vec![PushItem::new("orders", json!({ "orderId": 1001, "customer": 42, "total": 59.9 }))
           .partition("customer-42")      // one ordered partition per customer, created by this push
           .transaction_id("order-1001")]) // the dedup key: pushing the same id again is a no-op
       .await?;

   PushPHP

   $queen->queue('orders')
       ->partition('customer-42')        // one ordered partition per customer, created by this push
       ->push([[
           'transactionId' => 'order-1001', // the dedup key: pushing the same id again is a no-op
           'data' => ['orderId' => 1001, 'customer' => 42, 'total' => 59.9],
       ]])
       ->execute();

   PushC++

   client.queue("orders")
       .partition("customer-42") // one ordered partition per customer, created by the push
       .push({{
           {"transactionId", "order-1001"},  // the dedup key: pushing it again is a no-op
           {"data", {{"orderId", 1001}, {"customer", 42}, {"total", 59.9}}},
       }});

   Partitions →Deduplication →

4. 4

   Consume

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

   ConsumeJavaScript

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

   ConsumePython

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

   ConsumeGo

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

   ConsumeRust

   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?;

   ConsumePHP

   $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();

   ConsumeC++

   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.

   Consume with a group, every option setJavaScript

   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' })
     })

   Consume with a group, every option setPython

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

   Consume with a group, every option setGo

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

   Consume with a group, every option setRust

   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?;

   Consume with a group, every option setPHP

   $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();

   Consume with a group, every option setC++

   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.

   Groups, leases and acks →

5. 5

   One step, one commit

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

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

   Ack, KV, push and timer in one transactionJavaScript

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

   Ack, KV, push and timer in one transactionPython

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

   Ack, KV, push and timer in one transactionGo

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

   Ack, KV, push and timer in one transactionRust

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

   Ack, KV, push and timer in one transactionPHP

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

   Ack, KV, push and timer in one transactionC++

   client.queue("orders")
       .group("fulfillment")
       .subscription_mode("all")
       .auto_ack(false)
       .each()
       .limit(1)
       .consume([&](const json& msg) {
           const json& order = msg["data"];
           const std::string id = std::to_string(order["orderId"].get<int>());
           queen::TimerSchedule reminder;                       // a message tomorrow
           reminder.queue = "reminders";
           reminder.timer_key = "review-" + id;
           reminder.delay_millis = 24 * 60 * 60 * 1000;
           reminder.payload = {{"orderId", order["orderId"]},
                               {"customer", order["customer"]}};
   
           // client.transaction() cannot give its push a partition, so the step goes
           // to the broker's transaction route through the client's own HTTP layer.
           json step = {
               {"operations", json::array({
                   {{"type", "ack"}, {"status", "completed"},   // the input is done
                    {"consumerGroup", "fulfillment"}, {"leaseId", msg["leaseId"]},
                    {"transactionId", msg["transactionId"]},
                    {"partitionId", msg["partitionId"]}},
                   {{"type", "push"}, {"items", json::array({{  // the next event
                       {"queue", "shipments"}, {"partition", msg["partition"]},
                       {"transactionId", "ship-" + id},
                       {"payload", {{"orderId", order["orderId"]}}},
                   }})}},
               })},
               {"requiredLeases", json::array({msg["leaseId"]})},
               {"kv", json::array({queen::wire::kv_write_op("put", "orders", "order-" + id,
                   {{"status", "paid"}}, queen::KvTtl::seconds(7 * 86400), {})})},
               {"timers", json::array({queen::wire::timer_schedule_op(reminder)})},
           };
           client.get_http_client()->post("/api/v1/transaction", step);
       });

   Transactions →KV →Timers →

6. 6

   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.

   A new group, from a timeJavaScript

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

   A new group, from a timePython

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

   A new group, from a timeGo

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

   A new group, from a timeRust

   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?;

   A new group, from a timePHP

   $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();

   A new group, from a timeC++

   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.

   Fetch a partition by offsetJavaScript

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

   Fetch a partition by offsetPython

   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"])

   Fetch a partition by offsetGo

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

   Fetch a partition by offsetRust

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

   Fetch a partition by offsetPHP

   $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";
   }

   Fetch a partition by offsetC++

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

   Where a new group starts →Reads by offset →

7. 7

   Run a job once

   Some work no message triggers: a nightly report, a cleanup, a singleton. Take a lock for it. A lock is a lease, and a slow holder can outlive it, so guard the transaction with it: the push commits only while the lock is still yours.

   A guarded jobJavaScript

   const lock = queen.lock('daily-report', { ttl: '30s' }) // a lease with a fencing token
   if (await lock.acquire()) {                             // false: another worker has it
     try {
       await queen.transaction()
         .guard(lock)                                      // commits only while the lock is ours
         .queue('reports').push([{ transactionId: 'report-2026-10-22', data: { day: '2026-10-22' } }])
         .commit()
     } finally {
       await lock.release()
     }
   }

   A guarded jobPython

   lock = queen.lock("daily-report", ttl_seconds=30)  # a lease with a fencing token
   if await lock.acquire():                           # False: another worker has it
       try:
           await (
               queen.transaction()
               .guard(lock)                           # commits only while the lock is ours
               .queue("reports")
               .push([{"transactionId": "report-2026-10-22", "data": {"day": "2026-10-22"}}])
               .commit()
           )
       finally:
           await lock.release()

   A guarded jobGo

   lock := client.Lock("daily-report", 30*time.Second) // a lease with a fencing token
   held, err := lock.TryAcquire(ctx)                   // false: another worker has it
   if err != nil {
   	log.Fatal(err)
   }
   if held {
   	_, err = client.Transaction().
   		Guard(lock). // commits only while the lock is ours
   		Queue("reports").
   		Push(queen.PushItem{TransactionID: "report-2026-10-22", Payload: map[string]interface{}{"day": "2026-10-22"}}).
   		Commit(ctx)
   	_, _ = lock.Release(ctx)
   	if err != nil {
   		log.Fatal(err)
   	}
   }

   A guarded jobRust

   let lock = queen.lock("daily-report", Duration::from_secs(30)); // a lease with a fencing token
   if lock.acquire().await? {                                      // false: another worker has it
       let committed = queen
           .transaction()
           .guard(&lock) // commits only while the lock is ours
           .push_item(TxnPushItem {
               queue: "reports".into(),
               partition: None,
               payload: json!({ "day": "2026-10-22" }),
               transaction_id: Some("report-2026-10-22".into()),
               trace_id: None,
           })?
           .commit()
           .await;
       lock.release().await?;
       committed?;
   }

   A guarded jobPHP

   $lock = $queen->lock('daily-report', ['ttlSeconds' => 30]); // a lease with a fencing token
   if ($lock->acquire()) {                                    // false: another worker has it
       try {
           $queen->transaction()
               ->guard($lock)                                 // commits only while the lock is ours
               ->queue('reports')
               ->push([['transactionId' => 'report-2026-10-22', 'data' => ['day' => '2026-10-22']]])
               ->commit();
       } finally {
           $lock->release();
       }
   }

   A guarded jobC++

   queen::Lock lock = client.lock("daily-report", 30);  // a lease with a fencing token
   if (lock.acquire()) {                                // false: another worker has it
       try {
           client.transaction()
               .guard(lock)                   // commits only while the lock is ours
               .queue("reports")
               .push({{{"transactionId", "report-2026-10-22"},
                       {"data", {{"day", "2026-10-22"}}}}})
               .commit();
       } catch (...) {
           lock.release();                    // given back when the commit throws, too
           throw;
       }
       lock.release();
   }

   Locks and semaphores →

8. 8

   Aggregate with a stream

   A stream reads a queue and writes its results to another. The partition is the key, so each customer gets its own windows. Like a new group it starts at the tail: push a few more orders and a result arrives a minute later.

   Revenue per customer per minuteJavaScript

   import { Stream } from 'queen-mq'
   
   const revenue = await Stream
     .from(queen.queue('orders'))                 // one window per partition: per customer
     .windowTumbling({ seconds: 60 })
     .aggregate({ count: () => 1, sum: (order) => order.total })
     .to(queen.queue('revenue-per-minute'))       // one result per customer per minute
     .run({ queryId: 'revenue-per-minute', url: 'http://localhost:6632' })

   Revenue per customer per minutePython

   from queen import Stream
   
   revenue = await (
       Stream.from_(queen.queue("orders"))           # one window per partition: per customer
       .window_tumbling(seconds=60)
       .aggregate({"count": lambda order: 1, "sum": lambda order: order["total"]})
       .to(queen.queue("revenue-per-minute"))        # one result per customer per minute
       .run(query_id="revenue-per-minute", url="http://localhost:6632")
   )

   Revenue per customer per minuteGo

   import "github.com/smartpricing/queen/clients/client-go/v2/streams"
   
   revenue, err := streams.
   	From(client.Queue("orders").AsStreamSource()). // one window per partition: per customer
   	WindowTumbling(60).
   	Aggregate(map[string]streams.ExtractorFn{
   		"count": func(order interface{}) (float64, error) { return 1, nil },
   		"sum": func(order interface{}) (float64, error) {
   			return order.(map[string]interface{})["total"].(float64), nil
   		},
   	}, "count", "sum").
   	To(client.Queue("revenue-per-minute")). // one result per customer per minute
   	Run(ctx, streams.RunOptions{QueryID: "revenue-per-minute", URL: "http://localhost:6632"})
   if err != nil {
   	log.Fatal(err)
   }

   Revenue per customer per minuteRust

   use queen_mq::streams::{RunOptions, Stream};
   
   let revenue = Stream::from(queen.queue("orders")) // one window per partition: per customer
       .window_tumbling(60)
       .aggregate_count("count")
       .aggregate_sum("sum", |order| order.number("total"))
       .to(queen.queue("revenue-per-minute")) // one result per customer per minute
       .run(&queen, RunOptions::new("revenue-per-minute"))
       .await?;

   PHP has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

   C++ has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

   Streams →

9. 9

   Ask and answer

   Ephemeral queues live in memory, outside the log, for request/reply. The worker answers on whatever inbox a request names; the caller makes a private inbox and waits on it. There is no transaction or replay here, and a crash loses what is in flight.

   WorkerJavaScript

   const { messages } = await queen.ephemeral.pop('price-requests', { group: 'pricers', wait: true })
   for (const m of messages) {
     await queen.ephemeral.push(m.payload.replyTo, [{ sku: m.payload.sku, price: 59.9 }])
   }
   await queen.ephemeral.ack('price-requests', messages, { group: 'pricers' })

   WorkerPython

   result = await queen.ephemeral.pop("price-requests", group="pricers", wait=True)
   for m in result["messages"]:
       await queen.ephemeral.push(m["payload"]["replyTo"], [{"sku": m["payload"]["sku"], "price": 59.9}])
   await queen.ephemeral.ack("price-requests", result["messages"], group="pricers")

   WorkerGo

   batch, err := client.Ephemeral().Pop(ctx, "price-requests",
   	queen.EphemeralPopOptions{Group: "pricers", Wait: true})
   if err != nil {
   	log.Fatal(err)
   }
   for _, m := range batch.Messages {
   	var req struct{ Sku, ReplyTo string }
   	_ = json.Unmarshal(m.Payload, &req)
   	_, _ = client.Ephemeral().Push(ctx, req.ReplyTo,
   		[]map[string]interface{}{{"sku": req.Sku, "price": 59.9}})
   }
   _, _ = client.Ephemeral().Ack(ctx, "price-requests", batch.Messages,
   	queen.EphemeralAckOptions{Group: "pricers"})

   WorkerRust

   let eph = queen.ephemeral();
   let batch = eph.pop("price-requests").group("pricers").wait(true).send().await?;
   for m in &batch.messages {
       let reply_to = m.payload["replyTo"].as_str().unwrap_or_default();
       eph.push(reply_to, json!({ "sku": m.payload["sku"], "price": 59.9 })).send().await?;
   }
   eph.ack(&batch).await?;

   WorkerPHP

   $result = $queen->ephemeral()->pop('price-requests', ['group' => 'pricers', 'wait' => true]);
   foreach ($result['messages'] as $m) {
       $queen->ephemeral()->push($m['payload']['replyTo'], [['sku' => $m['payload']['sku'], 'price' => 59.9]]);
   }
   $queen->ephemeral()->ack('price-requests', $result['messages'], ['group' => 'pricers']);

   WorkerC++

   queen::EphemeralPopOptions take;
   take.group = "pricers";
   take.wait = true;
   json batch = client.ephemeral().pop("price-requests", take);
   for (const json& m : batch["messages"]) {
       json answer = {{"sku", m["payload"]["sku"]}, {"price", 59.9}};
       client.ephemeral().push(m["payload"]["replyTo"], json::array({answer}));
   }
   queen::EphemeralAckOptions done;
   done.group = "pricers";
   client.ephemeral().ack("price-requests", batch["messages"], done);

   CallerJavaScript

   const inbox = `inbox-${crypto.randomUUID()}`           // a private reply queue
   await queen.ephemeral.push('price-requests', [{ sku: 'A-1', replyTo: inbox }])
   const reply = await queen.ephemeral.pop(inbox, { wait: true, timeout: 5000, commitOnDelivery: true })
   console.log(reply.messages[0]?.payload)                // { sku: 'A-1', price: 59.9 }

   CallerPython

   inbox = f"inbox-{uuid.uuid4()}"                # a private reply queue
   await queen.ephemeral.push("price-requests", [{"sku": "A-1", "replyTo": inbox}])
   reply = await queen.ephemeral.pop(inbox, wait=True, timeout=5000, commit_on_delivery=True)
   print(reply["messages"][0]["payload"])         # {'sku': 'A-1', 'price': 59.9}

   CallerGo

   inbox := "inbox-" + queen.GenerateUUID() // a private reply queue
   _, err = client.Ephemeral().Push(ctx, "price-requests",
   	[]map[string]interface{}{{"sku": "A-1", "replyTo": inbox}})
   if err != nil {
   	log.Fatal(err)
   }
   reply, err := client.Ephemeral().Pop(ctx, inbox,
   	queen.EphemeralPopOptions{Wait: true, TimeoutMillis: 5000, CommitOnDelivery: true})
   if err != nil {
   	log.Fatal(err)
   }
   fmt.Println(string(reply.Messages[0].Payload)) // {"price":59.9,"sku":"A-1"}

   CallerRust

   let eph = queen.ephemeral();
   let inbox = format!("inbox-{}", queen_mq::uuid::uuidv7()); // a private reply queue
   eph.push("price-requests", json!({ "sku": "A-1", "replyTo": inbox })).send().await?;
   let reply = eph.pop(&inbox).wait(true).poll_timeout(Duration::from_secs(5))
       .commit_on_delivery(true).send().await?;
   println!("{}", reply.messages[0].payload); // {"price":59.9,"sku":"A-1"}

   CallerPHP

   $inbox = 'inbox-' . bin2hex(random_bytes(8));      // a private reply queue
   $queen->ephemeral()->push('price-requests', [['sku' => 'A-1', 'replyTo' => $inbox]]);
   $reply = $queen->ephemeral()->pop($inbox, ['wait' => true, 'timeout' => 5000, 'commitOnDelivery' => true]);
   print_r($reply['messages'][0]['payload']);          // [sku => A-1, price => 59.9]

   CallerC++

   auto inbox = "inbox-" + queen::util::generate_uuid_v7();      // a private reply queue
   json request = {{"sku", "A-1"}, {"replyTo", inbox}};
   client.ephemeral().push("price-requests", json::array({request}));
   queen::EphemeralPopOptions reply_wait;
   reply_wait.wait = true;
   reply_wait.timeout_millis = 5000;
   reply_wait.commit_on_delivery = true;
   json reply = client.ephemeral().pop(inbox, reply_wait);
   std::cout << reply["messages"][0]["payload"] << "\n";      // {"price":59.9,"sku":"A-1"}

   Ephemeral queues →

Push, consume, ack, replay, fetch, transactions, KV, timers, locks, streams and ephemeral queues all combine. Build on them and Queen carries the logic you would otherwise write around a broker.

The program behind these snippets: examples/tutorials/js/00-five-minutes.mjs

The program behind these snippets: examples/tutorials/py/00_five_minutes.py

The program behind these snippets: examples/tutorials/go/five-minutes/main.go

The program behind these snippets: examples/tutorials/rust/src/bin/00_five_minutes.rs

The program behind these snippets: examples/tutorials/php/00-five-minutes.php

The program behind these snippets: examples/tutorials/cpp/00-five-minutes.cpp