Queues & Priority

Named queues, priority ordering, pause/resume, and per-queue limits.

Every job belongs to a queue (default "default"). Queues are routing labels in shared storage — a worker chooses which queues it serves, and jobs dequeue in priority order within each queue.

Named queues and routing

Route tasks to different queues to isolate workloads — separate I/O-bound tasks (API calls, emails) from CPU-bound ones (data processing, report generation), and run each on a dedicated worker process.

python

@queue.task(queue="emails")
def send_email(to, subject, body):
    ...

@queue.task()  # goes to the "default" queue
def process_data(data):
    ...

send_email.delay("user@example.com", "Welcome", "...")

queue.task("sendEmail", (to: string, subject: string, body: string) => {
  // ...
});

const id = queue.enqueue("sendEmail", [to, subject, body], { queue: "emails" });

java

Task<EmailPayload> sendEmail = Task.of("send_email", EmailPayload.class).queue("emails");

String id = flexiq.enqueue(sendEmail, payload);

rust

#[flexiq::task(queue = "emails")]
fn send_email(to: String, subject: String, body: String) -> flexiq::Outcome<()> {
    Ok(())
}

#[flexiq::task] // goes to the "default" queue
fn process_data(data: Vec<u8>) -> flexiq::Outcome<()> {
    Ok(())
}

queue.enqueue(send_email::call(to, subject, body))?;

// Or override the declared default for one call.
queue.enqueue(send_email::call(to, subject, body).queue("urgent-emails"))?;

Override the decorator's default at enqueue time:

python

send_email.apply_async(args=(to, subject, body), queue="urgent-emails")

Routing only happens at enqueue time — the options object passed to queue.task() has no queue field, so there's no per-task default to set or override; every enqueue() call picks the queue independently.

The queue can also be set per enqueue, overriding the task's default:

java

flexiq.enqueue(sendEmail, payload, EnqueueOptions.builder().queue("urgent-emails").build());

See enqueue options for the rest of what an enqueue call can override.

Two places set the queue and the later one wins: the queue attribute seeds every call(..) this task builds, and .queue(..) on one call replaces it for that job. A task with neither lands on "default".

A queue is a label on the row, not a thing you create — naming one is all it takes. That also means a typo is silent: "emial" is a perfectly good queue, and its jobs sit pending forever because no worker was told to serve it. Keep the names in one const and use it on both sides.

See enqueue options for the rest of what a call can override.

Worker queue subscription

A worker only dequeues jobs from the queues it's told to serve — queues it isn't watching accumulate jobs untouched, no matter how many pile up.

python

queue.run_worker(queues=["emails", "reports"])

queue.runWorker({ queues: ["emails", "default"] }); // serve two queues

java

flexiq.worker()
        .handle(sendEmail, p -> deliver(p))
        .queues("emails", "default")   // serve two queues
        .start();

rust

let worker = queue
    .worker()
    .register::<send_email>()
    .queues(["emails", "reports"]) // serve two queues
    .spawn()?;

Omit queues(..) and the worker serves "default" alone.

Or from the CLI:

bash

# Process only email tasks
flexiq worker --app myapp:queue --queues emails

# Process multiple queues
flexiq worker --app myapp:queue --queues emails,reports

# Process all registered queues (default)
flexiq worker --app myapp:queue

Or from the CLI:

bash

flexiq run ./app.js --queues emails,reports

There's no worker subcommand on the bundled CLI — workers are code, started with flexiq.worker(). The CLI does have pause/resume subcommands for queue control; see CLI.

There is no worker binary and no subcommand: a worker is code, and its queue list is the queues([..]) argument. Reading that list from the environment is ordinary Rust, which is how one binary serves a different set per deployment:

rust

let queues: Vec<String> = std::env::var("FLEXIQ_QUEUES")
    .unwrap_or_else(|_| "default".into())
    .split(',')
    .map(str::to_string)
    .collect();

queue.worker().register::<send_email>().queues(queues).spawn()?;

Use queues to isolate workloads

Separate I/O-bound tasks (API calls, emails) from CPU-bound tasks (data processing, report generation) into different queues. Run them on different worker processes for optimal resource usage.

Priority

Higher-priority jobs dequeue first within a queue. Priority is a plain integer with no fixed scale — pick whatever range fits your app.

python

# This specific job is extra urgent
urgent_task.apply_async(args=(data,), priority=100)  # ahead of priority=0 jobs

queue.enqueue("report", [id], { priority: 10 }); // ahead of priority 0 jobs

java

flexiq.enqueue(report, payload, EnqueueOptions.builder().priority(10).build());
// ahead of priority 0 jobs

rust

queue.enqueue(report::call(id).priority(10))?; // ahead of priority 0 jobs

A registration-time default is the priority attribute: #[flexiq::task(priority = 10)]. A per-call .priority(..) still wins.

Coming from BullMQ?

Priority direction is inverted. BullMQ treats a lower priority number as higher priority (1 runs before 10). flexiq is the other way round — a higher priority number dequeues first. Porting a BullMQ priority scheme means inverting the numbers, not copying them.

Priority is a plain i32 with no fixed scale and no clamping — negatives are fine, and -10 is simply behind 0. Ties break on the oldest eligible scheduled_at, so equal-priority work stays fair rather than arbitrary.

Default priority

Set a default at task registration; individual enqueues can still override it:

python

@queue.task(priority=10)
def urgent_task(data):
    ...

@queue.task(priority=0)  # default
def normal_task(data):
    ...

Set a default on the task descriptor; a per-enqueue EnqueueOptions.priority(...) still overrides it:

java

Task<OrderPayload> urgentTask = Task.of("urgent_task", OrderPayload.class).priority(10);

There's no registration-time default — priority is enqueue-only, set per call as shown above.

Set a default on the task; a per-call .priority(..) still overrides it:

rust

#[flexiq::task(priority = 10)]
fn urgent_task(data: String) -> flexiq::Outcome<()> { Ok(()) }

#[flexiq::task] // priority 0, the default
fn normal_task(data: String) -> flexiq::Outcome<()> { Ok(()) }

The attribute seeds the EnqueueOptions every call(..) starts from, so this is the same mechanism as the per-call setter rather than a second one.

How it works

Higher-priority jobs go first.
Among jobs of equal priority, the one that became eligible earliest (the oldest scheduled_at) goes first.
Each queue is ordered independently of the others.

Pause and resume

Pausing stops dispatch for a queue without stopping the worker — in-flight jobs finish, new ones wait. Paused queues still accept new enqueues; they just won't be dequeued until resumed.

python

queue.pause("emails")
queue.paused_queues()   # ["emails"]
queue.resume("emails")

queue.pauseQueue("emails");
queue.listPausedQueues(); // ["emails"]
queue.resumeQueue("emails");

java

Queue emails = flexiq.queue("emails");
emails.pause();
emails.isPaused();              // true
flexiq.listPausedQueues();     // ["emails"]
emails.resume();

Pause and resume are not on FlexiQ. They are Storage methods, so reach them through queue.storage():

rust

use flexiq::Storage;

queue.storage().pause_queue("emails")?;
queue.storage().list_paused_queues()?; // ["emails"]
queue.storage().resume_queue("emails")?;

See Job management for maintenance-window patterns, cancellation, and archival built on top of pause/resume.

Pausing is a storage-level flag, so it is not namespaced: pause_queue takes a queue name and nothing else, and every namespace sharing that backend stops dispatching from it. FlexiQ::with_namespace does not scope this.

A paused queue still accepts enqueues — queue.enqueue(..) succeeds and the row is written — it just is not dispatched until resume_queue. In-flight jobs are untouched and run to completion.

Per-queue limits

A rate limit or concurrency cap applied to an entire queue, independently of any per-task settings — checked in the scheduler before per-task limits, so it's the right knob for protecting a shared downstream resource (an API, a database) regardless of which task is hitting it.

python

queue.set_queue_rate_limit("emails", "20/s")   # max 20 emails per second
queue.set_queue_concurrency("reports", 2)      # heavy tasks: max 2 at a time

The format is the same as rate_limit on @queue.task(): "N/s", "N/m", or "N/h". Both settings are read once when run_worker() starts — call them beforehand, or between worker restarts to change them.

queue.configureQueue("emails", {
  rateLimit: "50/s",
  maxConcurrent: 10,
});

rateLimit follows the same "<count>/<unit>" spec as a task's rateLimit option. Limits are applied when a worker calls runWorker() — set them before starting the worker.

The Java SDK has no queue-level rate limit or concurrency cap — throughput is shaped per task instead: producer-side gates for rate limiting (see Rate limiting) and worker thread-pool sizing for concurrency (see Concurrency).

The limits exist in the scheduler — QueueConfig carries a rate_limit and a max_concurrent, and the poller checks both before the per-task caps — but WorkerBuilder does not expose the setter. It is flexiq_core::Worker::queue_config(name, QueueConfig { .. }), so reaching it means building the worker through flexiq_core::Worker directly, which costs the durable-step fence this shell's pool carries.

The per-task knobs are the ones this shell does expose, and the same scheduler enforces them across every worker on shared storage: #[flexiq::task(rate_limit = "20/s", max_concurrent = 2)]. What they cannot do is cover several tasks sharing one downstream with a single number.

Per-queue stats

Scope job counts to a single queue, or get every queue's counts in one call.

python

queue.stats_by_queue("emails")   # {"pending": 3, "running": 1, ...}
queue.stats_all_queues()         # {"emails": {...}, "reports": {...}}

await queue.statsByQueue("emails");
await queue.statsAllQueues();

java

long backlog = flexiq.statsByQueue("emails").pending;
Map<String, QueueStats> all = flexiq.statsAllQueues();

rust

use flexiq::Storage;

// `FlexiQ::stats()` covers the whole handle; the per-queue reads are on storage.
let emails = queue.storage().stats_by_queue("emails", queue.namespace())?;
let all = queue.storage().stats_all_queues(queue.namespace())?;

Both take the namespace explicitly, which is what queue.namespace() returns — None on an unscoped handle.

Queue-level defaults

Configure defaults for every task at the Queue level; individual @queue.task() decorators override them:

python

queue = Queue(
    db_path="myapp.db",
    default_priority=0,    # default priority for all tasks
    default_retry=3,       # default max retries
    default_timeout=300,   # default timeout in seconds
)

There's no queue-level defaults object. Retry and timeout defaults are set per task on queue.task(...) via maxRetries and timeoutMs; a per-call enqueue() override still wins. priority has no default — it's enqueue-only (see Default priority above). configureQueue() only sets a queue's rateLimit / maxConcurrent.

queue.task("resize", (id: string) => { /* ... */ }, {
  maxRetries: 3, // default retry budget for this task
  timeoutMs: 300_000, // default per-job timeout (ms)
});

Java has no queue-level defaults object — defaults live on the task descriptor, and a per-enqueue EnqueueOptions still overrides them:

java

Task<ResizePayload> resize = Task.of("resize", ResizePayload.class)
        .priority(0)                       // default priority
        .maxRetries(3)                     // default retry budget
        .timeout(Duration.ofSeconds(300)); // default per-job timeout

There is no queue-level defaults object. FlexiQ::open(..) takes a path and nothing else; defaults live on the task and a per-call setter still wins:

rust

#[flexiq::task(priority = 0, max_retries = 3, timeout = "5m")]
fn resize(id: String) -> flexiq::Outcome<()> {
    Ok(())
}

Those three are already the built-in defaults, so the attribute above changes nothing — EnqueueOptions::default() is queue "default", priority 0, three retries and a 300 000 ms timeout, and that is what seeds every call. Declare an attribute only where you want something else.