Queue & Background Jobs

Somebody signs up and you send them a welcome email. Your mail server takes four seconds. They sat and watched a loading spinner for four seconds, and if the mail server had been down they would have seen an error for something that worked.

The fix is not to do it during the request. You write down that the email needs sending, answer immediately, and something else does the work a moment later. That list of things-to-do is a queue, each entry on it is a job, and the separate process that picks them up and runs them is a worker.

queue.Dispatch(ctx, SendWelcomeEmail{UserID: user.ID})   // returns at once

Anything slow or failure-prone belongs here: email, image processing, calling somebody else’s API, nightly cleanups.

Quick start

The queue provider is already registered in new projects:

// bootstrap/providers.go
&queue.Provider{}

With no config, it auto-reads the database connection from your existing configs/database.go. Jobs are stored in the same database as your app.

Defining a Job

Jobs implement the Job interface:

package jobs

import (
    "context"
    "fmt"
    "github.com/lemmego/queue"
    "github.com/lemmego/tasker"
)

type SendWelcomeEmail struct {
    UserID int    `json:"user_id"`
    Email  string `json:"email"`
}

func (j *SendWelcomeEmail) Handle(ctx context.Context) error {
    fmt.Printf("Sending email to %s\n", j.Email)
    return nil
}

func init() {
    queue.RegisterJob("*jobs.SendWelcomeEmail", func() tasker.Job {
        return &SendWelcomeEmail{}
    })
}

Dispatching Jobs

queue.Dispatch(ctx, &SendWelcomeEmail{
    UserID: 42,
    Email:  "user@test.com",
})

// With options
queue.Dispatch(ctx, &SendWelcomeEmail{UserID: 43},
    tasker.OnQueue("email"),
    tasker.WithDelay(5*time.Minute),
    tasker.WithTags("priority"),
)

Running Workers

lemmego run tasker:work --queue=default,email --workers=3

Workers claim jobs from the database atomically and process them. Run multiple worker processes across different machines — they coordinate through the shared database.

Optional Job Interfaces

Customize job behavior by implementing optional interfaces:

Retry & Backoff

By default, jobs use exponential backoff: 2^attempt * 2s, max 24h. Max 3 attempts. Configure per job:

func (j *MyJob) MaxAttempts() int { return 5 }

Failed jobs appear on the dashboard with full error stack traces. Click “Retry” to re-queue them.

Worker Pools

Configure pools per queue:

&queue.Provider{
    Config: &queue.Config{
        Workers: map[string]int{
            "default": 3,
            "email":   5,
            "images":  10,
        },
    },
}

Enable auto-scaling to dynamically adjust workers based on queue backlog:

Config: &queue.Config{
    EnableAutoscale: true,
}

Scheduling Recurring Jobs

scheduler.New(mgr).Register(scheduler.ScheduledJob{
    ID:       "daily-report",
    Schedule: "0 6 * * *",  // every day at 6 AM
    Job:      &GenerateReport{},
    Queue:    "default",
})

Web Dashboard

The dashboard lives at http://localhost:8080/tasker/, and it is closed until you say who may use it. Visiting it before then answers 401.

That is deliberate rather than an oversight. The dashboard can retry, cancel and delete jobs and pause a queue, so the failure mode of an open default is a public button that clears somebody’s work.

&queue.Provider{
    DashboardAuth: func(c app.Context) bool {
        if err := auth.Check(c); err != nil {
            return false
        }
        user, ok := auth.UserAs[*models.User](c)
        return ok && slices.Contains(taskerAdmins(), user.Email)
    },
}

Return true to let the request through, false for 401.

In a scaffolded project this is already wired, reading a comma-separated list from TASKER_ADMINS:

TASKER_ADMINS=ada@example.com,grace@example.com

auth.Check resolves a session cookie and a bearer token alike, so the same rule covers a browser and an API client — see The Current User. And nothing obliges the rule to involve a user at all; a shared secret header or an IP allowlist is just as valid, which is why DashboardAuth is a predicate over the request rather than over a user:

DashboardAuth: func(c app.Context) bool {
    return subtle.ConstantTimeCompare(
        []byte(c.Header("X-Tasker-Key")), []byte(key)) == 1
},

CSRF protection is enabled alongside it automatically, since the dashboard posts from a browser.

Once open:

API Endpoints

All endpoints are mounted under /tasker/, and every one of them is behind DashboardAuth — including the POSTs that retry, cancel and prune:

Configuration

Via provider options

&queue.Provider{
    Config: &queue.Config{
        Driver:             "sql",         // or "redis"
        DSN:                "postgres://...",
        DriverName:         "postgres",
        TablePrefix:        "tasker_",
        RoutePrefix:        "/tasker",
        DefaultQueue:       "default",
        DefaultMaxAttempts: 3,
        Workers:            map[string]int{"default": 3},
    },
}

Via .env

A scaffolded project gets internal/configs/tasker.go, which reads these. Before the queue had a config file, every TASKER_* variable was ignored and the queue ran on its built-in defaults.

TASKER_ROUTE_PREFIX=/jobs
TASKER_ADMINS=ada@example.com,grace@example.com
TASKER_TABLE_PREFIX=myapp_
TASKER_QUEUE=default
TASKER_MAX_ATTEMPTS=3
TASKER_HEARTBEAT_SEC=5
TASKER_REQUEUE_SEC=60
TASKER_PRUNE_HOURS=168
TASKER_AUTOSCALE=false

Which database the queue uses

With the SQL driver and no TASKER_DSN, the queue uses the application’s own connection, so jobs live in the same database as everything else and there is one thing to back up.

Point it somewhere else with TASKER_DSN if you want the queue isolated — but be careful what you point it at. A queue on an in-memory database accepts jobs and loses every one of them on restart, silently: the tables are recreated empty and nothing reports that anything was dropped. The queue logs a warning when it detects one.

Driver Support

SQL (PostgreSQL / MySQL / SQLite)

Default driver. Stores jobs in a tasker_jobs table with atomic claiming using SELECT ... FOR UPDATE SKIP LOCKED. Supports all SQL databases.

Redis

Alternative driver for node coordination, leader election, and distributed locking (job queue implementation uses database).

Architecture

   ┌──────────────┐     ┌──────────────┐
   │   Web App    │     │  Worker 1    │
   │  Dispatch    │     │  Supervisor  │
   │  Dashboard   │     │  └─Pool (3)  │
   └──────┬───────┘     └──────┬───────┘
          │                    │
          └────────┬───────────┘
                   │
          ┌────────▼────────┐
          │   PostgreSQL    │
          │  (jobs table)   │
          │  SELECT FOR     │
          │  UPDATE SKIP    │
          │  LOCKED         │
          └─────────────────┘

Multiple workers share the same database — they claim available jobs atomically. No direct inter-node communication needed. Scale by adding more worker processes.

CLI Commands

lemmego run tasker:work            # Start worker
  --queue=default,email            #   Queues to listen on
  --workers=3                      #   Workers per queue

lemmego run tasker:dispatch        # Dispatch test jobs
  --count=10
  --queue=default
  --fail-rate=30                   # % chance of failure
  --delay=500                      # Simulated processing delay (ms)

lemmego run tasker:job retry 5     # Retry job #5
lemmego run tasker:job cancel 5    # Cancel job #5

lemmego run tasker:queue list      # List queue stats