Queue

Background work shares the rest of the framework’s shape. A queue handler is a method on any #[injectable] provider, tagged #[process(queue = <Name>Queue, retries)] inside a #[processor] impl block. Importing RedisWorkerModule in the worker app activates the runtime at boot, drains the discovered methods, and spawns one worker per method. A producer-only app skips that import and only gets the connection — same crate, same types, one less module.

A queue is named by a type, not a string. #[queue(name = "...", job = <Payload>)] on a unit struct at the feature port declares the wire name and the payload in one artifact both sides import, so a typo or a mismatched payload is a compile error rather than a job that silently never drains.

Read this section, then start from the working reference: the worker app in Publish — audio transcoding jobs, with the api app as the producer.

The abstractions — Job, Processor, ProcessMethod, the #[processor] macro — ship in nest-rs-queue. The first-class storage, Redis, ships in nest-rs-redis (built on apalis): the RedisQueueProducer producer binding, the RedisWorker transport, and the RedisQueueModule / RedisWorkerModule activation seams. Both are compilation units, not lines you write: the redis feature below pulls them together, and an app’s manifest names neither.

Install

cargo add nest-rs --features redis

One line in the feature crate’s manifest: redis implies queue — the Redis-bound types, the macros, the worker-context seam and the pipe fold #[process] expands to, together.

A job payload declares serde itself; anyhow and tracing come through nest_rs::core::anyhow and nest_rs::queue::tracing. nestrs g queue <feature> writes the stanza for you.

The queue’s identity

One declaration, at the feature port — beside the payload, not inside the queue/ adapter. The producer is usually the port’s own service, so this is the artifact both sides reach for:

crates/features/src/audio/command.rs (from the demo)

use nest_rs::queue::{QueueName, queue};
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TranscodeCommand {
    pub file: String,
}

#[queue(name = "audio", job = TranscodeCommand)]
pub struct AudioQueue;

pub const AUDIO_QUEUE: &str = <AudioQueue as QueueName>::NAME;

Re-export both from the feature’s mod.rs. A marker that is not reachable from the port leaves the untyped push(name, job) escape hatch as the only way to enqueue — which loses the check QueueName exists to provide.

A first queue handler

A regular #[injectable] service with one decorated method. The #[processor] attribute marks the impl block; #[process] marks the method as a queue consumer.

crates/features/src/audio/queue/processor.rs (from the demo)

use std::sync::Arc;

use nest_rs::core::anyhow::Result;
use nest_rs::core::injectable;
use nest_rs::queue::processor;

use crate::audio::{AudioQueue, AudioService, TranscodeCommand};

#[injectable]
pub struct AudioProcessor {
    #[inject]
    svc: Arc<AudioService>,
}

#[processor]
impl AudioProcessor {
    #[process(queue = AudioQueue, retries = 3)]
    async fn transcode(&self, job: TranscodeCommand) -> Result<()> {
        self.svc.transcode(&job.file).await
    }
}

The processor is thin: it owns no business logic of its own — it hands job.file to AudioService::transcode and propagates the result. The same AudioService is what an HTTP producer calls to enqueue, so a single audited choke point describes everything the feature can do. Adapters carry transport mechanics; the service carries intent.

• #[injectable] makes AudioProcessor a regular DI provider.
• #[processor] on the impl block orchestrates the per-method #[process] attributes and submits one queue consumer per decorated method.
• #[process(queue, retries)] binds the method to a QueueName type and gives it a retry budget. The job type is read from the method’s first parameter after &self, and the macro asserts it is AudioQueue::Job — a handler consuming the wrong payload is a compile error naming both types.

The job argument also binds a pipe: Piped<P, T> / Valid<T> expose the wire payload T, run the pipe after deserialization, and hand the handler the transformed value — a rejection is a job error, so the queue’s retry/failure policy applies.

One job at a time

A #[process] method processes one job at a time, and there is no knob to change that. Throughput comes from running more replicas of the worker app:

$ kubectl scale deploy/worker --replicas=6   # six jobs in flight, not one

That is a deliberate division of labour. nestrs targets the container, and the container platform already schedules, meters and restarts the unit that decides how much work runs at once. A per-method ceiling would be a second scheduler competing with the first, and the number that makes it correct depends on the pod’s CPU share — something the source cannot know. Serialized-per-method is instead a property you can read off the code: every replica carries an identical load, and a handler that guards a connection pool or a rate-limited third-party API needs no ceiling of its own to stay within its budget.

Two consequences worth knowing:

• A slow handler is head-of-line blocking for its own queue — but only its own. Methods on other queues, and other replicas on the same queue, keep draining. Split a long tail onto its own queue rather than reaching for parallelism inside one.
• The consumer prefetches nothing. Each poll fetches a single job, so a job waiting behind a busy worker stays in Redis where a peer replica can take it — scaling out actually redistributes work instead of shuffling it between in-memory buffers.

Caution

Write #[process] handlers to be idempotent. Two replicas never receive the same job from the queue — the Redis fetch claims ids atomically — but starting a replica while another holds a job puts that job back on the queue, and it runs again. Scaling the worker up mid-flight therefore re-runs in-flight work, not only failed work. The measurement and the cause are in Retries and failure.

One attempt, one transaction

A job has no caller — system work, intentionally — so no ability is installed and Repo reads are unscoped, which is correct for work with no principal to scope to.

The executor is a transaction per attempt: it opens on the job’s first Repo call, commits when the method returns Ok and rolls back otherwise, so an attempt that failed halfway leaves nothing for the retry to write again. Add transactional = false to the decorator — #[process(queue = AudioQueue, transactional = false)], and the same word on #[every], #[cron] and #[after] — to run on the pool instead, which is right for a job that brackets long work that is not the database’s. Such a job owns its own idempotency.

The transaction bounds what a retry repeats, never what a redelivery does: a durable backend delivers at least once, so a worker that dies between the commit and the acknowledgement runs a job whose writes already landed. See Repo and the executor and Retries and failure.

Several methods on the same provider

The pattern the framework is built for: one #[injectable] declares the deps once, multiple decorated methods share them. Different queues, same service:

Three queues, so three #[queue] declarations at the port — each pairing a wire name with the payload it carries:

crates/features/src/audio/command.rs

#[queue(name = "audio", job = TranscodeCommand)]
pub struct AudioQueue;

#[queue(name = "audio.preview", job = PreviewCommand)]
pub struct AudioPreviewQueue;

#[queue(name = "audio.cleanup", job = CleanupCommand)]
pub struct AudioCleanupQueue;

crates/features/src/audio/queue/processor.rs (from the demo)

#[injectable]
pub struct AudioProcessor {
    #[inject] audio_svc: Arc<AudioService>,
    #[inject] media_svc: Arc<MediaService>,
}

#[processor]
impl AudioProcessor {
    #[process(queue = AudioQueue, retries = 3)]
    async fn transcode(&self, job: TranscodeCommand) -> Result<()> {
        self.audio_svc.transcode(&job.file).await
    }

    #[process(queue = AudioPreviewQueue, retries = 1)]
    async fn preview(&self, job: PreviewCommand) -> Result<()> {
        self.media_svc.preview(&job.file).await
    }

    #[process(queue = AudioCleanupQueue)]
    async fn cleanup(&self, job: CleanupCommand) -> Result<()> {
        self.media_svc.cleanup(job.older_than).await
    }
}

#[process(queue = AudioCleanupQueue)] omits the optional retries (default 0).

Three inventory entries — AudioProcessor::transcode, AudioProcessor::preview, AudioProcessor::cleanup — all pointing at the same AudioProcessor instance. Same Arc<AudioService>, same Arc<MediaService>, three queues. Each method still delegates: the processor is a router, never a place where business logic accumulates.

Caution

A single #[processor] impl block per struct (multiple would conflict at the discovery layer). Put every queue handler of a service in the same block — same convention as #[routes] and #[scheduled].

Wire it in

Three modules, three roles:

• RedisModule::for_root(None) — opens the one Redis connection every Redis binding shares (NESTRS_REDIS__URL). Every app that touches Redis needs it.
• RedisQueueModule — binds the producer over that connection, as Arc<dyn JobProducer>. Every app that enqueues imports it.
• RedisWorkerModule — activates the consumer runtime. At boot the framework drains the discovered #[process] methods, spawns one apalis worker per method, and shuts them down with the app. A producer-only app does NOT import this.

apps/worker/src/module.rs (from the demo)

use nest_rs::config::ConfigModule;
use nest_rs::core::module;
use nest_rs::redis::{RedisModule, RedisQueueModule, RedisWorkerModule};

use features::audio::AudioQueueModule;

#[module(imports = [
    ConfigModule::for_root(),
    RedisModule::for_root(None),
    RedisQueueModule,
    RedisWorkerModule::for_root(None),
    AudioQueueModule,
])]
pub struct WorkerModule;

AudioQueueModule imports the AudioModule port itself, so the app lists only the adapter — registration is idempotent.

main is the universal skeleton:

apps/worker/src/main.rs (from the demo)

use nest_rs::core::anyhow::Result;
use nest_rs::core::App;

use worker::WorkerModule;

#[tokio::main]
async fn main() -> Result<()> {
    App::builder()
        .module::<WorkerModule>()
        .build()
        .await?
        .run()
        .await
}

Run it

$ nestrs run dev worker

 INFO nest_rs::app: attached module-contributed transport transport="RedisWorker"
 INFO nest_rs::queue: registered queue processor processor="AudioProcessor::transcode" queue="audio" retries=3
 INFO features::audio: transcoded file="track-1787069801760.mp3" byte_size=50 trace_id=01a015a9252076e399f736a97ae90784 span_id=7bbcfc44c1f0c676
 INFO nest_rs::operation: queue.job queue="audio" processor="AudioProcessor::transcode" job_id="01M0ATJ9C2YN8NJX3GVTA911X6" attempt=1 outcome="ok" duration_ms=33.733 trace_id=01a015a9252076e399f736a97ae90784 span_id=7bbcfc44c1f0c676

The last two lines share the job’s trace_id, and it is the producer’s — see Observability for the full field set.

In this section

Basics

• Producing jobs — typing payloads, the publisher handle, and the rule that keeps push_to::<Q> behind a service.
• Wiring — RedisModule, RedisQueueModule and RedisWorkerModule, and module-gating across a shared features crate.

All options

• Retries and failure — what an Err means, the apalis retry shape, panics, dead-lettering.
• Observability — the span, its fields, and the lines a worker emits per job.
• Writing a queue driver — the scaffold for a non-Redis backend.

Reference

• crates/nest-rs-queue/ — the abstractions: #[processor], #[process], Job, Processor, ProcessMethod, JobProducer.
• crates/nest-rs-redis/ — the Redis storage: RedisModule and its RedisConnection, RedisQueueModule and its RedisQueueProducer, RedisWorkerModule and its RedisWorker.

Going further

• Scheduling — recurring work driven by a clock rather than a producer.
• Events — in-process fan-out, when the work need not outlive the request.
• Correlation — the trace an enqueue hands to the attempt, across the process boundary.