seppv0.1.0
Workers

Processing jobs

Write a worker: handlers, retries, concurrency, lease auto-extend, idempotency, graceful shutdown and the raw reserve/ack/nack loop.

A worker reserves jobs, runs them and reports how each one went. The Worker does all of that plumbing for you: reserving, leasing, acking and nacking, running jobs concurrently, and shutting down cleanly. You just supply the handlers. This page covers writing a worker with Worker, then drops down to the raw reserve/ack/nack primitives for when you need to drive the loop yourself.

The snippets assume a connected client. See connecting to set one up.

Define a worker

Give the Worker a client, the queues to pull from and a lease duration, then register a handler for each job type. It pulls from every queue you list and routes each job to the handler for its type. A handler receives the job's payload and a ctx with its id, attempt, custom metadata and more; returning normally acknowledges the job. Call run and the worker reserves and dispatches jobs until it's shut down.

use std::time::Duration;
use sepp_rs::client::SeppClient;
use sepp_rs::worker::Worker;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = SeppClient::connect("http://localhost:50051").await?;

    let worker = Worker::new(client, ["emails", "checkout"], Duration::from_secs(30))?
        .handle("welcome", |payload, ctx| async move {
            println!("processing job {} (attempt {})", ctx.id, ctx.attempt);
            if let Some(p) = &payload {
                println!("  payload: {} bytes ({})", p.data.len(), p.encoding);
            }
            // Do the work here. Returning Ok acks the job.
            Ok(())
        })?
        .handle("charge_card", |_payload, ctx| async move {
            println!("charging card for job {}", ctx.id);
            // Do the work here.
            Ok(())
        })?;

    worker.run().await;
    Ok(())
}

Report success or failure

A handler that returns normally acks the job. To fail it, raise a HandlerError (in Rust, return Err(HandlerError::…)); that nacks the job and tells the server what to do next:

  • retry — back to the queue for another attempt, timed by the queue's retry policy, immediate by default (up to max_attempts, after which it's dead-lettered).
  • retry-after — the same, but not before the given delay.
  • permanent — skip retries and dead-letter it now.

In Python and TypeScript, any other exception that escapes the handler is treated as a retry, so a bug never silently drops a job.

use std::time::Duration;
use sepp_rs::worker::HandlerError;

// The handler passed to `.handle(...)`:
|payload, ctx| async move {
    // A validation failure can't be fixed by retrying — dead-letter it.
    let Some(payload) = payload else {
        return Err(HandlerError::permanent("missing payload"));
    };

    // `send_email` is your code; map its failures to a directive.
    match send_email(&payload.data).await {
        Ok(()) => Ok(()),
        Err(err) if err.is_rate_limit() => {
            Err(HandlerError::retry_after("rate limited", Duration::from_secs(60)))
        }
        Err(err) => Err(HandlerError::retry(err.to_string())),
    }
}

Concurrency and long-running jobs

By default a worker runs up to 16 jobs at once; raise or lower that with max-in-flight. If a handler might run longer than the lease, turn on auto-extend and the worker renews the lease in the background for as long as the handler is still running — so you never have to extend it by hand.

Because any job can be retried or redelivered, the handlers should be safe to run more than once. See idempotent handlers.

Use auto-extend with care

Auto-extend can keep a lease alive indefinitely if the handler hangs or never finishes. That is why it is opt-in and why a handler under it should have its own timeout. See extend a lease manually to renew at explicit checkpoints instead.

let worker = Worker::new(client, ["emails"], Duration::from_secs(30))?
    .with_max_in_flight(64) // up to 64 jobs at once (default 16)
    .with_auto_extend()     // renew the lease while a handler is still running
    .handle("welcome", |payload, ctx| async move {
        // ...
        Ok(())
    })?;

Extend a lease manually

Auto-extend renews on a fixed heartbeat. To renew at explicit points instead — between the chunks of a long job, say — call ctx.extend from the handler yourself. It pushes the lease out from now and returns the new expiry, clamped to the queue max; a failed extend means the lease was already lost, which the worker treats as a retry.

use std::time::Duration;
use sepp_rs::worker::HandlerError;

.handle("import", |payload, ctx| async move {
    for chunk in chunks(&payload) {
        process_chunk(chunk).await;
        // Renew the lease between chunks so a long job keeps it.
        ctx.extend(Duration::from_secs(30))
            .await
            .map_err(|e| HandlerError::retry(e.to_string()))?;
    }
    Ok(())
})?

Idempotent handlers

At-least-once delivery means a handler can run more than once for the same job: a redelivery after a lost lease is indistinguishable from a first run and concurrent workers can briefly hold the same job. So a handler's side effects must be safe to repeat.

This is the handler-side counterpart to producer-side enqueue deduplication. The dedup key stops duplicate jobs; an idempotent handler stops one job's duplicate effects. You usually want both. Practical techniques:

  • Key writes on something stable (the job id or a business key carried in custom) and use a conditional insert or upsert, so a second run is a no-op rather than a double-write.
  • Make outbound calls idempotent where the API supports it (an idempotency key on the payment, a PUT rather than a POST).
  • Use ctx.attempt and ctx.max_attempts when the final attempt needs different handling, for example writing to a dead-letter table of your own.
  • Keep the time between committing side effects and acking the job as short as possible, so the window for a completed job getting redelivered is as small as possible.

A job that can never succeed (malformed input, a deleted record) is a poison job: left alone it burns every attempt before dead-lettering. Detect those and fail them with a permanent nack so they skip straight to dead-letter instead of retrying. See retries & dead-letter.

Shut down gracefully

Workers run forever, so you need a clean way to stop — especially on deploys, where in-flight jobs should finish rather than be dropped and redelivered. Take a shutdown handle before run, trigger it from a signal, and run stops reserving new jobs, waits for the in-flight ones to drain and returns.

let shutdown = worker.shutdown_handle();
tokio::spawn(async move {
    tokio::signal::ctrl_c().await.ok();
    shutdown.shutdown(); // stop reserving; let in-flight jobs finish
});

worker.run().await; // returns once draining is complete

The primitives: reserve, ack, nack, extend

The Worker is built on four calls you can use directly when you need control it doesn't expose. reserve long-polls a set of queues and hands back leased jobs; ack, nack and extend report on a job you hold.

reserve takes the queues to pull from and a lease duration. It returns as soon as a job is ready or after the wait timeout with nothing if the queues stay empty. Queues are tried in listed order, so put the most important first. You get a single job by default; ask for a batch with max_jobs. The server clamps wait_timeout, lease_duration and max_jobs to its advertised maxima rather than erroring, so a larger request is fine.

use std::time::Duration;
use sepp_rs::ReserveOptions;

// Pull up to 10 jobs, "emails" first then "checkout", each leased for 30s.
// Wait up to 5s for the first job to show up.
let opts = ReserveOptions::new(["emails", "checkout"], Duration::from_secs(30))?
    .with_max_jobs(10)
    .with_wait_timeout(Duration::from_secs(5));

// Some(jobs) holds at least one job; None means the wait elapsed with
// nothing ready, so you just reserve again.
if let Some(jobs) = client.reserve(&opts).await? {
    for job in &jobs {
        println!("leased {} (attempt {})", job.ctx.id, job.ctx.attempt);
    }
}

When you ack, nack or extend, the call can come back as a lost lease: the job already moved to another worker or it is gone entirely. The clients surface this as JobNotFound and AttemptMismatch (both under LeaseError). Treat both the same way — you no longer own the job, so stop rather than reprocess or retry the call. See lease expiry and fencing for why.

use sepp_rs::client::LeaseError;

// After processing, ack the job you hold.
match client.ack(&job.ctx).await {
    Ok(()) => {} // the job is gone for good
    // The lease was lost; don't reprocess it.
    Err(LeaseError::JobNotFound | LeaseError::AttemptMismatch) => {
        eprintln!("lost the lease on {}, skipping", job.ctx.id);
    }
    Err(LeaseError::Client(e)) => return Err(e.into()),
}

Hand-roll the loop

The Worker is the right tool almost always. But if you need control it doesn't expose — custom reserve batching, your own concurrency model, bespoke bookkeeping — drive the primitives yourself: reserve a batch, process each job, and ack or nack it. Here the outcome is a RetryDirective passed to nack rather than a HandlerError. A production loop also needs error backoff, lease-loss handling and draining on shutdown, which the Worker gives you for free.

use std::time::Duration;
use sepp_rs::client::{RetryDirective, SeppClient};
use sepp_rs::{Job, ReserveOptions};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = SeppClient::connect("http://localhost:50051").await?;
    let opts = ReserveOptions::new(["emails"], Duration::from_secs(30))?;

    loop {
        let Some(jobs) = client.reserve(&opts).await? else {
            continue; // nothing ready yet, poll again
        };

        for job in jobs {
            match handle(&job).await {
                Ok(()) => {
                    client.ack(&job.ctx).await?;
                }
                Err(err) => {
                    client.nack(&job.ctx, RetryDirective::Default, err.to_string()).await?;
                }
            }
        }
    }
}

async fn handle(job: &Job) -> Result<(), Box<dyn std::error::Error>> {
    println!("processing job {} ({})", job.ctx.id, job.ctx.job_type);
    // Do the work here.
    Ok(())
}

See also

Enqueue from the admin UI

Produce jobs from the browser: the admin UI's enqueue form and how it maps onto EnqueueRequest.

Client SDKs

First-party SDKs (Rust, Python, JavaScript/TypeScript).

On this page

Define a workerReport success or failureConcurrency and long-running jobsExtend a lease manuallyIdempotent handlersShut down gracefullyThe primitives: reserve, ack, nack, extendHand-roll the loopSee also