reliable

reliable moves each entry to a processing list and removes it on ack, which makes the queue at-least-once.

The subscription claims an entry with BLMOVE, which moves it from the queue to the processing list in one command. The entry stays there while the handler runs. An ack removes it with LREM. A retry pushes it back onto the queue with LPUSH, behind the entries already waiting, and then removes it from the processing list. A crash between those two commands leaves the entry in both lists, so it is handled twice rather than lost. A drop removes it, as an ack does. A consumer that dies after the move leaves its entry on the processing list; a recovery set returns such entries to the queue.

Example

The render farm turns the first job away while it warms up, so that job goes back on the queue and is rendered when it is popped again.

Macros
Manual

examples/redis_lists.rs
/// The entry waits on a processing list while the handler runs: an ack removes it, a retry puts
/// it back onto `renders`, behind the jobs waiting there.
#[subscriber(RedisList::new("renders").reliable())]
async fn render(job: &Job, State(farm): State<RenderFarm>) -> HandlerOutcome {
    match farm.render(job).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(job.id, %err, "render failed, back on the queue");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/redis_lists.rs
/// The entry waits on a processing list while the handler runs: an ack removes it, a retry puts
/// it back onto `renders`, behind the jobs waiting there.
struct Render;

impl Handle<Job, (), (), (), AppState> for Render {
    async fn handle(
        &self,
        job: &Job,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        if let Err(err) = ctx.state().farm.render(job).await {
            tracing::warn!(job.id, %err, "render failed, back on the queue");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn renders() -> RedisList {
    RedisList::new("renders").reliable()
}

Example​

Example