A queue is created by its first push. Configure it for leases, retries, delays, deduplication, retention and encryption, with the default and the reason for each option.

A queue is a named set of partitions that share one configuration. You never have to create one: the first push that names a queue creates it, with the defaults. Configure it when you want other settings.

The queue orders holds its options (leaseTime, retryLimit, retryDelay, dedupWindowSeconds, retention), shared by all its partitions. Its partitions are customer-42, customer-7 and customer-13, each an ordered list of messages at offsets 0, 1, 2 and so on. A push that names a new partition, customer-99, creates it. — A queue: options shared by its partitions, each an ordered list of messages.

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

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

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

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

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

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

Why

a queue created by a push is good enough to start, and most queues never need more. When they do, the settings live on the queue, in one place, instead of in every worker.

Every option

Here is a queue that sets most of them, and the table below says what each one does.

await queen.queue('jobs')
  .config({
    leaseTime: 120,                   // 2 minutes per job, then it goes to another worker
    retryLimit: 5,                    // after 5 failures, the dead-letter queue
    retryDelay: 10000,                // a failed job comes back after 10 s, not at once
    delayedProcessing: 30,            // a new job waits 30 s before anyone can take it
    windowBuffer: 5,                  // a burst waits 5 s, then is handed over at once
    dedupWindowSeconds: 86400,        // a transactionId is remembered for a day
    retentionEnabled: true,           // retention runs only when this is on:
    completedRetentionSeconds: 3600,  //   jobs every group has finished go after an hour
    maxWaitTimeSeconds: 604800,       // anything older than a week goes, finished or not
  })
  .create()

await queen.queue("jobs").config({
    "leaseTime": 120,                   # 2 minutes per job, then it goes to another worker
    "retryLimit": 5,                    # after 5 failures, the dead-letter queue
    "retryDelay": 10000,                # a failed job comes back after 10 s, not at once
    "delayedProcessing": 30,            # a new job waits 30 s before anyone can take it
    "windowBuffer": 5,                  # a burst waits 5 s, then is handed over at once
    "dedupWindowSeconds": 86400,        # a transactionId is remembered for a day
    "retentionEnabled": True,           # retention runs only when this is on:
    "completedRetentionSeconds": 3600,  #   jobs every group has finished go after an hour
    "maxWaitTimeSeconds": 604800,       # anything older than a week goes, finished or not
}).create()

_, err = client.Queue("jobs").
	Config(queen.QueueConfig{
		LeaseTime:                 120,    // 2 minutes per job, then another worker
		RetryLimit:                5,      // after 5 failures, the dead-letter queue
		RetryDelayMillis:          10000,  // a failed job is back after 10 s, not at once
		DelayedProcessing:         30,     // a new job waits 30 s before anyone takes it
		WindowBuffer:              5,      // a burst waits 5 s, then is handed over at once
		DedupWindowSeconds:        86400,  // a transactionId is remembered for a day
		RetentionEnabled:          true,   // retention runs only when this is on:
		CompletedRetentionSeconds: 3600,   //   jobs all groups finished go after an hour
		MaxWaitTimeSeconds:        604800, // anything over a week old goes, finished or not
	}).
	Create().
	Execute(ctx)
if err != nil {
	log.Fatal(err)
}

let options = QueueOptions {
    lease_time: Some(120),                  // 2 minutes per job, then to another worker
    retry_limit: Some(5),                   // after 5 failures, the dead-letter queue
    retry_delay: Some(10000),               // a failed job returns after 10 s, not at once
    delayed_processing: Some(30),           // a new job waits 30 s before anyone takes it
    window_buffer: Some(5),                 // a burst waits 5 s, then goes out as one
    dedup_window_seconds: Some(86400),      // a transactionId is remembered for a day
    retention_enabled: Some(true),          // retention runs only when this is on:
    completed_retention_seconds: Some(3600), //   jobs all groups finished go after an hour
    max_wait_time_seconds: Some(604800),    // older than a week: gone, finished or not
    ..Default::default()
};
queen.queue("jobs").configure(options).await?;

$queen->queue('jobs')->config([
    'leaseTime' => 120,                  // 2 minutes per job, then another worker gets it
    'retryLimit' => 5,                   // after 5 failures, the dead-letter queue
    'retryDelay' => 10000,               // a failed job comes back after 10 s, not at once
    'delayedProcessing' => 30,           // a new job waits 30 s before anyone can take it
    'windowBuffer' => 5,                 // a burst waits 5 s, then is handed over at once
    'dedupWindowSeconds' => 86400,       // a transactionId is remembered for a day
    'retentionEnabled' => true,          // retention runs only when this is on:
    'completedRetentionSeconds' => 3600, //   jobs every group has finished go after an hour
    'maxWaitTimeSeconds' => 604800,      // anything older than a week goes, finished or not
])->create()->execute();

// QueueConfig has none of retryDelay, dedupWindowSeconds, retentionEnabled and
// maxWaitTimeSeconds, so the options go to the broker's configure route
// through the client's own HTTP layer.
json options = {
    {"leaseTime", 120},                  // 2 minutes per job, then another worker
    {"retryLimit", 5},                   // after 5 failures, the dead-letter queue
    {"retryDelay", 10000},               // a failed job returns after 10 s, not at once
    {"delayedProcessing", 30},           // a new job waits 30 s before anyone takes it
    {"windowBuffer", 5},                 // a burst waits 5 s, then goes out at once
    {"dedupWindowSeconds", 86400},       // a transactionId is remembered for a day
    {"retentionEnabled", true},          // retention runs only when this is on:
    {"completedRetentionSeconds", 3600}, //   jobs all groups finished go after an hour
    {"maxWaitTimeSeconds", 604800},      // older than a week goes, finished or not
};
client.get_http_client()->post("/api/v1/configure",
                               {{"queue", "jobs"}, {"options", options}});

Option	Default	What it does
`leaseTime`	60 s (300 s from `create()`)	How long a worker holds the messages it took before they go to another worker.
`retryLimit`	3	How many failed attempts a message gets. After the last one it goes to the dead-letter queue.
`retryDelay`	1000 ms	How long a failed message waits before it comes back. Its partition waits with it, for that consumer group.
`deadLetterQueue`, `dlqAfterMaxRetries`	true	File a message that ran out of retries in the dead-letter queue. With both off, it is dropped.
`delayedProcessing`	0 s	How long every new message waits after its push before anyone can take it.
`windowBuffer`	0 s	How long new messages wait in their partition, so a burst is handed over as one batch.
`dedupWindowSeconds`	3600 s	How long a partition remembers `transactionId`s. `0` turns deduplication off.
`retentionEnabled`	false	Must be true for the two options below to do anything.
`retentionSeconds`	0	Remove messages older than this, consumed or not.
`completedRetentionSeconds`	0	Remove messages older than this that every consumer group has already acked.
`maxWaitTimeSeconds`	0	Remove messages older than this, consumed or not. Works without `retentionEnabled`.
`encryptionEnabled`	false	Encrypt payloads on disk. Every node needs `QUEEN_ENCRYPTION_KEY`.
`namespace`, `task`	from the name	Labels that group queues in the dashboard and in a discovery pop.

Leases. A worker that takes messages holds them under a lease. Until it acks them or the lease runs out, no other worker of the same consumer group gets that partition.

Why

a worker can crash in the middle of its work. The lease is how its messages get back to someone else without a human noticing. Set it longer than your slowest handler, or renew it while the handler runs (see Pop messages).

Retries. A failed message comes back retryLimit times, each time after retryDelay, and then goes to the dead-letter queue.

Why

most failures are temporary: a database that restarts, a provider that times out. Trying again later fixes them. Waiting between tries gives the other side time to recover, and a message that can never succeed stops after a few tries instead of blocking its partition forever.

Delays. delayedProcessing holds every new message of the queue for a while. windowBuffer holds a group’s new messages until the burst is over, so the handler gets them together.

Why

some work should not start at once (a reminder sent five minutes after a signup), and some work is cheaper in bulk (one database write for a burst of fifty updates). For a delay that is different for each message, use a timer instead (see KV, Locks and Timers).

Deduplication. Inside the dedup window, a second push with the same transactionId in the same partition is not written.

Why

producers retry when they do not get an answer, and a retry must not create a second order.

Retention. By default Queen keeps every message, and the disk grows. Retention runs only when retentionEnabled is true; then retentionSeconds removes old messages, read or not, and completedRetentionSeconds removes old messages that every group has finished. maxWaitTimeSeconds removes old messages without needing retentionEnabled. A node whose disk is 85% full refuses the writes that would grow it.

Why

an acked message is not deleted: it stays in its partition so that a new consumer group, an audit or a fixed bug can read it again. Retention is how you decide how long that history is worth its disk.

Encryption. With encryptionEnabled, payloads are encrypted on disk.

Why

some payloads hold personal data that must not sit on a disk in clear text.

Three traps

config(...).create() in the JavaScript client sends your options together with the client’s own defaults (a 300-second lease, encryption off, no retention). Calling it on an existing queue resets every option you leave out. Name every option you care about.
On a node without QUEEN_ENCRYPTION_KEY, an encrypted queue stores plain text, and nothing warns you.
priority, ttl, maxSize and minPopWaitTime are accepted and stored, and do nothing.
The C++ client cannot set retentionEnabled, retryDelay, dedupWindowSeconds or maxWaitTimeSeconds, so its code above posts to /api/v1/configure itself. A C++ queue configured with retention_seconds alone never runs retention.

Create and configure queues: every option

Every option

Three traps

See also