Queue

Background work shares the rest of the framework’s shape. A queue handler is a method on any #[injectable] provider, tagged #[process(queue = "...", concurrency, retries)] inside a #[processor] impl block. Importing QueueWorkerModule 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.

There is no step-by-step tutorial for the queue yet. Read this section, then start from the working reference: the worker app in Publish (audio transcoding jobs) and its api 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 QueueConnection producer, the QueueWorker transport, and the QueueModule / QueueWorkerModule activation seams. An app depends on both: nest-rs-queue for the macro and abstractions, nest-rs-redis for the Redis-bound types.

Install

cargo add nest-rs-queue nest-rs-redis

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.

use std::sync::Arc;

use anyhow::Result;
use nest_rs_core::injectable;
use nest_rs_queue::processor;

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

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

#[processor]
impl AudioProcessor {
    #[process(queue = "audio", concurrency = 5, 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, concurrency, retries)] binds the method to a Redis queue name (stringly-typed, the known cost), bounds the in-flight jobs, and gives an apalis retry budget. The job type is read from the method’s first parameter after &self.

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.

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, different concurrencies, same service:

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

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

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

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

#[process(queue = "audio.cleanup")] omits the optional concurrency (default 1) and 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.

Wire it in

Two modules, two roles:

QueueModule::for_root(None) — provides the shared QueueConnection. Every app (producer and consumer) needs it.
QueueWorkerModule — 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.

use nest_rs_config::ConfigModule;
use nest_rs_core::module;
use nest_rs_redis::{QueueModule, QueueWorkerModule};

use features::audio::AudioQueueModule;

#[module(imports = [
    ConfigModule::for_root(),
    QueueModule::for_root(None),
    QueueWorkerModule,
    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:

use 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::transport: attached module-contributed transport transport=QueueWorker
 INFO nest_rs::queue: registered queue processor processor=AudioProcessor::transcode queue=audio concurrency=5 retries=3
 INFO process job{queue=audio processor=AudioProcessor::transcode job_id=01HFEX… attempt=1}: features::audio: transcoded file=track-1717405126521.mp3
 INFO process job{queue=audio processor=AudioProcessor::transcode job_id=01HFEX… attempt=1}: nest_rs::queue: job ok elapsed_ms=342

The two process job{…} lines share the per-job span; see Observability for the full field set.

Going further

Producing jobs — typing payloads, the publisher handle, and the rule that keeps queue.of() behind a service.
Wiring — QueueModule vs QueueWorkerModule, producer-only deploys, and module-gating across shared features.
Retries and failure — what an Err means, the apalis retry shape, panics, dead-lettering.
Observability — span structure, fields, and expected logs.
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: QueueConnection, QueueWorker, QueueModule, QueueWorkerModule.