Job queue (battery/queue)

battery/queue is a pluggable job queue with three backends: in-memory, SQL (SQLite + Postgres), and Redis. It handles enqueue, dequeue, retry with optional exponential backoff, dead-letter capture, inspection, and replay, and pairs with a Scheduler for recurring jobs.

Backends at a glance

Pick MemoryQueue for tests and single-process prototypes. Use DBQueue when you need durability and multi-replica safety (Postgres FOR UPDATE SKIP LOCKED). Use RedisQueue when you already run Redis and want the visibility-timeout model.

Quickstart: MemoryQueue

import "github.com/DonaldMurillo/gofastr/battery/queue"

q := queue.NewMemoryQueue(4, queue.WithLogger(logger)) // 4 workers, log failures

q.RegisterHandler("send-email", func(ctx context.Context, job queue.Job) error {
    return sendEmail(ctx, job.Payload)
})

q.Start()
defer q.Close()

_ = q.Enqueue(ctx, queue.Job{
    Type:    "send-email",
    Payload: json.RawMessage(`{"to":"user@example.com"}`),
})

Quickstart: DBQueue

db, _ := sql.Open("postgres", dsn)

q, err := queue.NewDBQueue(db,
    queue.WithWorkers(4),
    queue.WithLeaseTimeout(2*time.Minute),
    queue.WithBackoff(5*time.Second, 5*time.Minute),
    queue.WithDBHandlerTimeout(30*time.Second), // cancel a stuck handler's ctx
    queue.WithDBLogger(logger),                 // route failures to your logger
)
if err != nil {
    log.Fatal(err)
}

q.RegisterHandler("process-upload", func(ctx context.Context, job queue.Job) error {
    return processUpload(ctx, job.Payload)
})

q.Start(ctx)
defer q.Close()

WithLeaseTimeout (default 5 minutes) is how long a claimed job may run before another worker treats its worker as dead and claims it again. Keep it longer than your slowest handler. A zero or negative value keeps the default, as does the same value passed to SetLeaseTimeout. The handler budget (WithDBHandlerTimeout) must be shorter than the lease: NewDBQueue refuses a pair where it is not, since a handler still inside its budget would be claimed and run a second time, and SetLeaseTimeout ignores such a lease and logs a WARN.

NewDBQueue creates the queue_jobs table and its index if they do not exist. Pass WithTable("my_jobs") to use a custom table name; it is exported and erased under that name like the default (App.ExportData, App.EraseUserData). An index name longer than Postgres's 63-byte limit ends in a hash of the table name, so a long table name still gets each of its indexes. The durable scheduler's tables and indexes (<table>_scheduler_schedules, _occurrences, _leases) follow the same rule on Postgres when the cut Postgres would make collides with another name; a cut that collides with nothing keeps that name, the one earlier releases created.

Quickstart: RedisQueue

// client implements queue.RedisClient; wrap go-redis, redigo, etc.
q := queue.NewRedisQueue(client, "myapp:jobs")
q.SetVisibilityTimeout(30 * time.Second)

// Launch the auto-reclaim ticker (re-delivers crashed-worker jobs).
q.Start(ctx, 30*time.Second)

// Enqueue
_ = q.Enqueue(ctx, queue.Job{Type: "notify", Payload: payload})

// Dequeue + process manually (no built-in worker pool for Redis)
for {
    job, err := q.Dequeue(ctx)
    if errors.Is(err, queue.ErrNoJob) {
        time.Sleep(time.Second)
        continue
    }
    if err := handle(ctx, job); err != nil {
        _ = q.Nack(ctx, job)
    } else {
        _ = q.Ack(ctx, job)
    }
}

RedisQueue does not include a built-in worker loop; you drive Dequeue/Ack/Nack yourself, or integrate with a third-party pool. Call Start to enable the auto-reclaim ticker (see "Crash safety" below).

Job struct

type Job struct {
    ID          string          // auto-filled by Enqueue if empty
    OccurrenceID string          // stable durable-schedule tick identity; empty for ordinary jobs
    Type        string          // required: selects the handler
    Payload     json.RawMessage // arbitrary JSON for the handler
    Priority    int             // higher = dequeued first (DBQueue + MemoryQueue)
    Lane        string          // capacity-reservation lane; "" = default (see Lanes)
    Attempts    int             // incremented on each claim
    MaxAttempts int             // auto-defaults to 3; 0 means 3
    CreatedAt   time.Time       // auto-filled if zero
    ScheduledAt time.Time       // auto-filled to now; set to delay execution
    ClaimToken  string          // minted by Dequeue; fences this claim's Ack/Nack
    UserID      string          // who the payload is about; makes the row erasable
    UniqueKey   string          // at most one live job per (Type, UniqueKey); see Unique jobs
}

Scheduled jobs (future ScheduledAt) are invisible to Dequeue until the moment passes, on every backend, and a due job enqueued behind one is still handed out. This lets you implement delayed processing without a separate scheduler. MemoryQueue workers sleep until the earliest held-back job comes due; Close does not wait for jobs that are not due yet, and drops them with a WARN log naming the count. RedisQueue keeps them on a <queue>:delayed list and moves the due ones onto the main list at the start of each Dequeue; that list is read whole on every call, so it suits hundreds of delayed jobs, not millions. A due job is pushed onto the main list before it leaves the delayed one, so a worker that dies mid-move leaves a duplicate (within the at-least-once contract), never a lost job.

UserID and the right to be forgotten

queue_jobs lives outside the entity registry, so App.EraseUserData can only reach a row through a declared column — and that column is user_id, populated from Job.UserID. Set it whenever the payload is personal data. A job that renders someone's address, points at their document, or carries their email keeps that data in the row, and App.ExportData dumps the payload column verbatim.

Terminal rows make this sharp: the only job-row DELETE is Ack-of-claimed, and failed rows are retained on purpose so Stats/ListJobs/Replay can see them. A dead-lettered job with no UserID therefore holds that payload indefinitely and re-discloses it in every later export.

Leave it empty for infrastructure work — cache warms, sweeps, index rebuilds — where there is no subject to erase.

RedisQueue and MemoryQueue are erased too: each constructor registers a store eraser (see Stores outside the database) that removes every job whose UserID is the erased user after the SQL erasure commits. For RedisQueue that is the main, delayed and dead lists and the pop and processing hashes; for MemoryQueue, the pending jobs (delayed ones included, and jobs a disabled module's gate holds back), the jobs Dequeue handed out, and the dead list, with the unique keys they held freed. The Redis erasure is one Lua script, so it needs an adapter that implements ScriptRunner; with one that does not, the erasure fails with an error and removes nothing. The MemoryQueue erasure holds the queue still while it runs: a Dequeue, Nack or Replay waits for it, so none carries a job from a store the erasure has not read yet into one it already has. The in-memory Scheduler is erased before the store-erased queues it feeds, so a tick cannot enqueue the user's job into them again. A DBQueue's rows go in the SQL erasure, which commits before any store eraser runs, so a tick in between can write the user's job back; the scheduler's eraser therefore deletes the user's rows from every DBQueue it feeds once its schedules are gone, and counts them in its report line. A worker still running an erased job finds nothing to settle, and its Ack or Nack does nothing. The registration stays until Close, and it keeps the queue from being garbage-collected, so close a queue you are done with. Two RedisQueues with one queue name on different servers are both erased.

A schedule sets UserID on the jobs it fires with its UserID option; see Scheduler.

Unique jobs

Set Job.UniqueKey when a second copy of a job is wasted or harmful: a per-user digest, a reindex of one record, a sync for one account. While a job with the same Type and key is live (pending, claimed, or waiting on a retry), Enqueue refuses another with queue.ErrDuplicateJob and enqueues nothing:

err := q.Enqueue(ctx, queue.Job{
    Type:      "digest",
    UniqueKey: "user:" + userID,
    Payload:   payload,
})
if err != nil && !errors.Is(err, queue.ErrDuplicateJob) {
    return err
}

The key is free again once the job succeeds or is dead-lettered, so the next Enqueue with that key goes through. A dead-lettered job keeps its key on the row, and Replay returns ErrDuplicateJob instead of reviving it while a newer job holds the key. An empty key never deduplicates, and the same key under two types is two keys.

On DBQueue a partial unique index on (type, unique_key) over pending and claimed rows holds the key, so it holds across replicas and inside a transaction. A duplicate skips the insert with ON CONFLICT DO NOTHING instead of raising, which keeps the caller's Postgres transaction usable. Tables created before the column existed gain it, and the index, when NewDBQueue runs.

RedisQueue returns queue.ErrUniqueUnsupported for a job with a key. RedisClient has no atomic set-if-absent to hold the key with, and a job enqueued without one would run as a duplicate.

Uniqueness is about enqueueing. A job still runs at least once, and a handler that a lease expiry re-runs is the duplicate the at-least-once contract already covers.

Lanes

A lane is a capacity-reservation tag on a job (Job.Lane; empty string is the default lane). Type still selects the handler; Lane only affects which workers can claim the job. It solves a problem priority cannot.

Why priority is not enough

Priority chooses among pending jobs when a worker frees up. It cannot preempt a running handler. So if a bulk backfill saturates every shared worker with long-running jobs, an urgent job, even at the highest priority, just sits in the queue: no worker is free to consult the priority order. Priority helps pick the next job; lanes guarantee there is always a worker whose entire budget is reserved for the jobs that matter.

Dedicated lane workers

WithDBLaneWorkers(lane, n) (DBQueue) and WithLaneWorkers(lane, n) (MemoryQueue) add n dedicated workers on top of the shared pool (WithWorkers / the NewMemoryQueue count) that only claim jobs whose Lane matches. Shared workers keep claiming any lane by priority.

q, _ := queue.NewDBQueue(db,
    queue.WithWorkers(4),                  // shared pool: claims any lane
    queue.WithDBLaneWorkers("high", 2),    // 2 workers reserved for the "high" lane
)
q.Start(ctx)

// A bulk backfill fills the shared pool…
_ = q.Enqueue(ctx, queue.Job{Type: "reindex", Lane: "bulk"})
// …but the high lane always has a free worker:
_ = q.Enqueue(ctx, queue.Job{Type: "send-alert", Lane: "high"})

Multiple calls for different lanes each add their own workers; multiple calls for the same lane sum. Pass a non-empty lane and n > 0 (both panic otherwise). Lease reclaim, backoff, the gate, and handler-timeout behaviour all apply identically to lane workers; they run the same claim loop with an extra AND lane = ? filter.

RedisQueue

RedisQueue has no worker loop of its own, so lane isolation is "one RedisQueue instance per lane" via its queueName: instantiate a queue per lane and dedicate workers to each. There is no Lane field consumed by the Redis backend.

Retry and backoff

By default, a Nack with attempts remaining makes the job immediately eligible again (next Dequeue can pick it up).

WithBackoff(base, max) turns on exponential backoff for DBQueue:

queue.WithBackoff(5*time.Second, 5*time.Minute)

The n-th retry delay is base × 2^(n-1), capped at max. A job that Nacks on attempt 1 waits ~5s; attempt 2 waits ~10s; attempt 3 waits ~20s; etc., up to 5m.

Once Attempts >= MaxAttempts, the job moves to the dead-letter state instead of being retried.

Dead-letter and replay

When a job exhausts MaxAttempts, it is retained as a terminally-failed job (never silently dropped):

• MemoryQueue: stored in a bounded in-memory slice (cap 1 000; oldest evicted on overflow).
• DBQueue: row status set to 'failed'.
• RedisQueue: appended to the <queue>:dead Redis list. Reclaim sends an unparseable processing-hash entry there too, and removes it from the hash only once the push landed: that hash holds a claimed job's only durable copy, so a corrupt entry is quarantined, never deleted.

MemoryQueue and DBQueue log every handler failure at WARN and every dead-letter at ERROR (via slog.Default(), or a custom logger passed as WithLogger / WithDBLogger); failures are no longer silent. (RedisQueue has no built-in worker loop, so it logs nothing itself; whatever drives Dequeue/Ack/Nack owns its own logging.)

Replay a failed job (reset attempts to 0 and re-enqueue):

// Type-assert the Replayable capability (all three backends implement it).
if r, ok := q.(queue.Replayable); ok {
    if err := r.Replay(ctx, jobID); err != nil {
        log.Printf("replay failed: %v", err)
    }
}

Replay is idempotent: replaying an unknown ID or a non-failed job is a no-op (returns nil, no side effect). Replaying a job whose UniqueKey a newer live job holds returns ErrDuplicateJob and leaves it failed.

Inspecting jobs (Browsable)

All three backends implement Browsable:

if b, ok := q.(queue.Browsable); ok {
    jobs, _ := b.ListJobs(ctx, "failed", 50, 0)
    stats, _ := b.Stats(ctx)
    fmt.Println("failed:", stats["failed"])
}

ListJobs accepts a status string ("pending", "failed", "" for all), a limit and an offset. Jobs are returned newest-first, after skipping the newest offset of them, so ListJobs(ctx, "failed", 50, 50) is the second page of 50; a negative offset reads as zero. Stats returns a JobStats map (status → count), which gives a pager its total.

MemoryQueue copies job payload bytes when enqueueing, tracking manual claims, returning dead-letter listings, and handing a job to its handler; editing a caller or handler payload cannot change a pending retry.

MemoryQueue and RedisQueue can only enumerate their dead-letter store, so only "failed", "all" or "" returns results. DBQueue can enumerate any status: "pending", "claimed" (running), "failed", and "done" with WithDoneRetention, and "all" or "" lists every status.

A job DBQueue lists carries Status, UpdatedAt (when its state last changed: enqueue, claim, retry, failure, replay or completion) and LastError, the error its last failed attempt returned through the worker, with control bytes scrubbed and cut to 500 runes. A replay keeps LastError.

By default Ack deletes a finished job. queue.WithDoneRetention(d) keeps it as "done" for d instead, so an admin can list what ran; the claim loop deletes done jobs older than d, at most once a minute. A kept job still holds its ID, so enqueuing the same ID again fails until the retention has passed. A Nack that arrives after the job is done, dead-lettered or replayed changes nothing.

Crash safety and auto-reclaim

DBQueue reclaims stale-claimed jobs automatically inside Dequeue: a row in claimed status whose claimed_at has passed the configured lease timeout (default 5 min) becomes eligible again. No extra configuration needed.

On shutdown (the context passed to Start cancelled), a DBQueue worker that has claimed a job but not started its handler hands the job back with its attempt refunded, so the next process runs it; a handler that is already running gets the cancelled context, and its result settles as usual. A worker that finds a claimed job's type gated hands it back the same way. A claim the database committed while the shutdown cancelled the query cannot be handed back; that job reruns when its lease expires.

RedisQueue uses a visibility timeout: while a job is in-flight it sits in a processing hash with an expiry timestamp. Call RedisQueue.Start(ctx, interval) to run an auto-reclaim ticker:

q.Start(ctx, 30*time.Second) // checks every 30 s; 0 defaults to 30 s, negative panics

The ticker calls q.Reclaim(ctx) on each tick, which scans the processing hash and re-enqueues any job whose expiresAt has passed. Without Start, crashed-worker jobs strand silently until you call Reclaim manually.

A worker that dies inside Dequeue, after the pop and before the processing write, leaves no processing entry to expire. With a client that implements PopRecorder, the pop leaves a record of the job in <queue>:popped, and Reclaim puts a record older than the visibility timeout back on the main list. Without it, that job is lost.

You can also call Reclaim directly from your own ticker:

n, err := q.Reclaim(ctx)
fmt.Printf("reclaimed %d jobs\n", n)

RedisQueue's Enqueue runs under a recover guard, so a panicking host-supplied RedisClient implementation surfaces as an ordinary error with a control-byte-scrubbed panic value, instead of unwinding the scheduler/dispatch loops. This applies to immediate and scheduled jobs. Claim fencing on the Redis backend compares ClaimToken values in constant time (subtle.ConstantTimeCompare), so match status is not leaked through timing.

Scheduler

In-memory, single-process mode

Scheduler enqueues recurring jobs with watermarks held only in process memory:

sched := queue.NewInMemoryScheduler(q) // NewScheduler remains a compatible alias

// Fixed interval: fires every 5 minutes.
sched.Every(5 * time.Minute).
    Job("send-digest", json.RawMessage(`{}`)).
    Register()

// Cron expression: fires every day at 02:00.
if err := sched.Cron("0 2 * * *").
    Job("nightly-rollup", nil).
    Register(); err != nil {
    log.Fatalf("bad cron spec: %v", err)
}

go sched.Start(ctx) // blocks until ctx is cancelled

// Lane / Priority / MaxAttempts on a schedule are carried verbatim into
// every Job the scheduler fires; set them to tag the fired job's lane
// (matching lane workers and any shared catch-all worker can claim it),
// to jump the dequeue order, or to bound retries per occurrence.
sched.Every(15 * time.Minute).
    Job("bulk-reindex", nil).
    Lane("bulk").         // bulk-lane AND shared workers can claim this
    Priority(-5).         // lower priority than ad-hoc work
    MaxAttempts(1).       // one shot: never retry a partial reindex
    Register()

Lane, Priority, and MaxAttempts are fluent options on both ScheduleBuilder and DurableScheduleBuilder. Omit them for today's defaults: empty lane, priority 0, and MaxAttempts resolved to 3 at enqueue time. They are carried unchanged into every Job the schedule fires. Lane("bulk") makes the fired job claimable by bulk-lane workers AND by any shared/catch-all worker; tagging a lane alone does NOT keep bulk work off interactive workers; to do that, dedicate workers to an interactive lane (so there is no shared pool draining bulk) instead of relying on the tag (see Lanes). Priority(n) nudges dequeue order among pending work, and MaxAttempts(k) bounds how many times a single occurrence may retry before dead-lettering. UserID(id) sets Job.UserID on every fired job, so App.EraseUserData reaches the jobs of a schedule whose payload is about one person (a per-user digest).

Every(d) schedules fire on a fixed interval of one millisecond or more (Register refuses a negative interval or a shorter positive one, the same floor as the durable scheduler; Every(0) fires on every tick of the one-minute floor); Cron(spec) schedules fire when the cron expression's next time arrives; use it for time-of-day work like "every day at 02:00" that an interval cannot express. A cron schedule is evaluated in the zone of the time it was registered at (Register uses the local zone, RegisterAt(t) uses t's) and handles daylight-saving changes like the durable scheduler below: a fixed-time spec runs once in a repeated hour and once, right after the jump, when a spring-forward skips its time. The spec is parsed by framework/cron (cron.Parse), so the queue does not carry a second cron parser; it accepts the same 5-field syntax and @shortcuts (e.g. @daily). The two kinds coexist in one scheduler.

Register() returns an error, and registers nothing, when the payload given to Job cannot be encoded as JSON (a NaN, a channel), or when a Cron spec is invalid or has no firing in the next five years (0 0 30 2 *, February 30, which DurableScheduler refuses too). A cron schedule whose next firing lies past that search stops and logs a warning (February 29 after 2096, since 2100 is not a leap year). RegisterAt(base) is the deterministic variant: it anchors the first run to base instead of time.Now(), which is handy for tests and replayed fixtures.

When the scheduler runs, the wake interval is the smallest of the interval schedules and one minute (cron resolution); a cron-only scheduler wakes once per minute. Jobs registered after Start still fire; the loop re-reads the schedule set each tick and a Register nudges it to re-arm immediately, so the natural "start subsystems, then register jobs" wiring works (it previously snapshotted once at Start and dropped everything registered later).

This mode is intentionally non-durable. A restart recomputes the first tick, and multiple scheduler replicas may enqueue the same tick. It remains useful for tests and applications that guarantee one scheduler process.

Durable, replica-safe mode

DurableScheduler requires DBQueue. It persists schedule definitions, watermarks, lease fences, and one unique occurrence per (schedule ID, scheduled tick) in the queue database:

q, err := queue.NewDBQueue(db)
if err != nil {
    log.Fatal(err)
}

durable, err := queue.NewDurableScheduler(q, queue.DurableSchedulerConfig{
    OwnerID:               hostname + ":" + instanceID,
    LeaseDuration:         30 * time.Second,
    OccurrenceRetention:   30 * 24 * time.Hour, // zero uses this default; negative disables pruning
    MaxCatchUpOccurrences: 1000,                // default; bounds history materialized after downtime
})
if err != nil {
    log.Fatal(err)
}

// The first argument is the stable schedule ID. Re-registering the same ID
// updates its definition without resetting the persisted next-run watermark.
if err := durable.Every("customer-digest", 5*time.Minute).
    Job("send-digest", json.RawMessage(`{}`)).
    Register(); err != nil {
    log.Fatal(err)
}
if err := durable.Cron("nightly-rollup", "0 2 * * *").
    Job("nightly-rollup", nil).
    Register(); err != nil {
    log.Fatal(err)
}

go func() {
    if err := durable.Start(ctx); err != nil {
        log.Printf("durable scheduler stopped: %v", err)
    }
}()

A durable Every takes one millisecond or more; Register refuses a shorter interval, and a stored row with one is skipped like a corrupted row.

Start survives evaluation errors: a corrupted schedule row (a bad next_run, a half-applied manual repair) is skipped with an ERROR log and the remaining due schedules still fire, and a transient database failure is logged and retried at the next heartbeat instead of ending the loop. The loop only exits when ctx is cancelled, so one poison row cannot stop every schedule on the replica. A skipped row is tried again at the next heartbeat (a third of LeaseDuration), not on every wake, so it logs once per heartbeat and does not keep the loop busy.

The same Lane / Priority / MaxAttempts options exist on the durable builder. They PERSIST alongside the schedule definition: re-registering the same schedule ID updates them without resetting the watermark, and every fired occurrence carries them into its Job (so a bulk lane stays routed to bulk workers across restarts and replicas):

if err := durable.Every("nightly-bulk-reindex", 24*time.Hour).
    Job("bulk-reindex", nil).
    Lane("bulk").
    Priority(-5).
    MaxAttempts(1).
    Register(); err != nil {
    log.Fatal(err)
}

The options columns (lane, priority, max_attempts, user_id) are added to an existing scheduler_schedules table by an idempotent migration during NewDurableScheduler; no manual schema work is required on upgrade.

A durable schedule with UserID(id) keeps its payload in the schedules table and fires it on every tick, so NewDurableScheduler registers that table with App.EraseUserData: erasing the user deletes the schedules that name them, and the queue_jobs eraser deletes the jobs they fired. A schedule registered before the option existed names nobody. The in-memory Scheduler registers a store eraser that drops the user's schedules from memory; Scheduler.Close removes it (it does not stop Start; cancel Start's context for that).

Cron field evaluation follows the location of the registration anchor. Register() anchors in UTC, so "0 2 * * *" fires at 02:00 UTC, the behavior schedules have always had. To evaluate a spec in a zone's wall-clock time (including across DST shifts), register with RegisterAt and a time in a named IANA zone:

loc, _ := time.LoadLocation("America/New_York")
if err := durable.Cron("nightly-rollup", "0 2 * * *").
    Job("nightly-rollup", nil).
    RegisterAt(time.Now().In(loc)); err != nil {
    log.Fatal(err)
}

Across DST shifts the schedule behaves like vixie cron. A fixed-time spec, one whose minute and hour fields both name values, fires once per wall-clock time. When clocks go back, a minute in the repeated hour fires at its first occurrence: "0,30 1 * * *" in New York fires at 01:00 and 01:30 EDT and not again at 01:00 or 01:30 EST, however the scheduler's heartbeats split the night. When clocks go forward, a minute the jump skips fires once, at the transition. A spec whose minute or hour field begins with * runs by the new wall clock: "*/30 * * * *" fires through both 01:00 hours, and "15 * * * *" does not fire for the 02:15 a spring-forward skips.

A spec that never fires ("0 0 30 2 *") is refused at registration. A rare one is not: "0 0 29 2 */7" (February 29 on a Sunday) next fires in 2032, and its gap across 2100 is 40 years.

The zone name persists with the schedule (a tz column added by the same idempotent migration as the options columns). UTC, fixed-offset zones (time.FixedZone), and time.Local store the empty default and evaluate in UTC, the first run included; time.Local would resolve to a different zone on every replica. The in-memory Scheduler differs here: it has no persistence, so its cron specs evaluate against the process clock in the process's local zone. Moving a schedule from Scheduler to DurableScheduler (Every/Cron … .Register()) on a non-UTC host shifts its fire time to the UTC wall clock; pass the intended zone via RegisterAt to keep local-time semantics. Re-registering a schedule with a different cadence keeps the stored watermark and self-heals: evaluation advances to the next valid occurrence of the new spec instead of failing. A schedule whose stored state cannot produce a due tick is logged and skipped each heartbeat; other schedules keep firing.

One replica holds a heartbeat-expiry lease for efficient evaluation. Every lease acquisition after expiry increments a fencing token. Before committing an occurrence, the scheduler re-checks that exact owner/token row. The occurrence insert, version-guarded watermark advance, and queue_jobs insert then commit in one SQL transaction. Each schedule has a monotonically increasing version; advancing its watermark uses WHERE id = ? AND version = ?, so timestamp precision or timezone normalization cannot make a valid compare-and-swap miss. A paused former owner therefore cannot enqueue after another replica reclaims the lease, and the unique occurrence identity is the deduplication authority even if leader evaluation races.

When evaluation wakes after several ticks, only the newest due tick is enqueued. Older retained ticks are stored as skipped occurrences instead of causing a catch-up burst. MaxCatchUpOccurrences bounds that materialized history window and defaults to 1,000: after a longer outage, the scheduler fast-forwards the watermark and retains only the newest bounded window; the walk over missed ticks never holds more than that many at once. This prevents a stale fixed-interval or cron watermark from allocating unbounded memory or creating an unbounded transaction. The enqueued Job.OccurrenceID is stable for that schedule ID and tick, so handlers and run-history records can correlate retries to the same occurrence. If the previous occurrence is still pending or claimed, the newest tick is also recorded with skip_reason = 'overlap' and no second job is enqueued.

Occurrence history is bounded automatically. OccurrenceRetention defaults to 30 days; zero selects that default and a negative duration disables automatic pruning. After evaluating due work, at most once per hour per scheduler process, the scheduler deletes old skipped rows and old enqueued rows only after their queue job is done or gone; live work and a failed job that may be replayed are never pruned, so a replay still blocks an overlapping tick. A newly elected replica may sweep immediately. The occurrence table includes (schedule_id, enqueued_job_id) and created_at indexes for overlap checks and retention sweeps. Set a longer retention when occurrence IDs feed an external audit or reconciliation process.

RunOnce(ctx, now) exposes deterministic/manual evaluation; Start(ctx) heartbeats and evaluates until cancellation.

Handler timeout. By default a DBQueue handler runs unbounded: a black-holed dependency (an SMTP host that never answers, a hung HTTP call) wedges the worker forever, and with the default single worker that stalls the whole queue. Pass WithDBHandlerTimeout(d) (WithHandlerTimeout(d) on MemoryQueue, 30s default there, kept for a zero or negative d; on DBQueue zero means no timeout) and the worker stops waiting at the deadline even if the handler ignores its cancelled context: the invocation runs on its own goroutine, the job is nacked at the deadline (retried/dead-lettered), and the worker keeps draining — so one non-cooperative handler costs one retry cycle, not the whole queue. A side effect landing after the deadline is the duplicate the at-least-once contract already requires handlers to tolerate. The bundled SMTP sender (battery/email) also bounds its own dial at 10s (SMTPConfig.DialTimeout), so it can't hang even without a handler timeout.

Multiple queues can be passed to NewScheduler; the job is enqueued onto all of them. Enqueue errors are logged via slog.Default(). NewSchedulerWithLogger lets you supply a custom *slog.Logger.

The in-memory Scheduler still fires without a distributed lock. Use DurableScheduler when more than one replica may evaluate schedules or when watermarks must survive restart.

Handler registration

Handlers are registered by job type. Unregistered types are acknowledged (dropped) so they never loop. Handlers are safe to register concurrently with a running worker loop.

q.RegisterHandler("resize-image", func(ctx context.Context, job queue.Job) error {
    // Return a non-nil error to Nack (retry or dead-letter).
    return resizeImage(ctx, job.Payload)
})

A handler panic is recovered and treated as an error; the job follows the normal retry path, so a poison message cannot drain the worker pool. If a panic escapes DBQueue's per-job guard (for example in claim or settlement plumbing), the worker logs an Error through its configured slog.Logger with its lane, scrubbed panic value and stack, then respawns after 100 ms. Queue shutdown or context cancellation interrupts that respawn delay.

Each handler call on MemoryQueue and DBQueue runs under an OpenTelemetry consumer span named process <job type>, carried by the handler's ctx. It is dropped unless the app installs a TracerProvider; see observability.

RedisClient interface

RedisQueue accepts any client that implements queue.RedisClient:

type RedisClient interface {
    LPush(ctx, key string, values ...interface{}) error
    RPop(ctx, key string) (string, error)
    HSet(ctx, key string, values ...interface{}) error
    HGet(ctx, key, field string) (string, error)
    HGetAll(ctx, key string) (map[string]string, error)
    HDel(ctx, key string, fields ...string) error
    Del(ctx, keys ...string) error
    LRange(ctx, key string, start, stop int64) ([]string, error)
    LRem(ctx, key string, count int64, value interface{}) (int64, error)
}

Wrap your preferred driver (go-redis, redigo, etc.) with a thin adapter that maps to this interface.

Optional: CompareAndDeleter

An adapter may additionally implement:

type CompareAndDeleter interface {
    HDelIfEqual(ctx, key, field, expect string) (deleted bool, err error)
}

Ack/Nack read the processing entry, compare its claim token, then delete it. Nothing binds that delete to the entry whose token was compared: one round trip of skew (a GC pause, a failover) is enough for the lease to expire, Reclaim to run, and another worker to re-claim — and the delete then removes the new claimant's entry, leaving the job on no list and invisible to Reclaim.

Implementing HDelIfEqual makes that delete atomic. With go-redis, a one-line Lua script:

if redis.call('HGET', KEYS[1], ARGV[1]) == ARGV[2] then
  return redis.call('HDEL', KEYS[1], ARGV[1]) end
return 0

Return false when the value had changed; the queue treats that as a fenced no-op and counts it in StaleClaimCount(). An adapter that does not implement it still works — RedisQueue falls back to re-reading immediately before the delete, which narrows the window but cannot close it.

Optional: CompareAndMover

type CompareAndMover interface {
    MoveIfEqual(ctx, key, field, expect, list, value string) (moved bool, err error)
}

Nack and Reclaim decide a requeue (a retry, a dead-letter, a lease-expiry re-delivery) from a snapshot of the processing entry, then push. HDelIfEqual cannot fence that: the push has already landed when the conditional delete refuses. If the lease expired and another worker re-claimed in between, the stale push puts a second copy on the main list, and a third worker claims it while the second still holds the job.

MoveIfEqual pushes value onto list and deletes the processing field only while the field still holds expect, in one step:

if redis.call('HGET', KEYS[1], ARGV[1]) == ARGV[2] then
  redis.call('LPUSH', KEYS[2], ARGV[3])
  redis.call('HDEL', KEYS[1], ARGV[1])
  return 1 end
return 0

On Redis Cluster a script may only touch keys in one slot: give the queue a hash-tagged name ({jobs}) so its lists and hash share one. Return false when the entry had changed; the queue counts it in StaleClaimCount() and pushes nothing. Without it, RedisQueue pushes and then releases with HDelIfEqual, which leaves that window open.

Optional: PopRecorder

type PopRecorder interface {
    RPopHSet(ctx, list, hash, field string) (string, error)
}

Dequeue pops a job and writes it to the processing hash in two round trips. A worker killed between them (a deploy, an OOM kill) runs no recovery write, and the job is on no list. RPopHSet pops the tail of list and stores the value in hash under field in one step:

local v = redis.call('RPOP', KEYS[1])
if not v then return false end
redis.call('HSET', KEYS[2], ARGV[1], v)
return v

The record's field holds the pop time. Dequeue deletes the record once the job has its next home: the processing hash, the dead-letter list, or back on the main list. Reclaim handles a record older than the visibility timeout: if the processing hash holds the job under the same claim, it drops the record, and otherwise it puts the job back on the main list. Report an empty list as queue.ErrRedisEmpty (go-redis returns redis.Nil for the script's false). Use the same hash-tagged queue name on Redis Cluster as for CompareAndMover.

Two cases deliver a job twice, which the at-least-once contract allows: a record whose delete failed, once its job has finished and left the processing hash, and a record that a Dequeue slower than the visibility timeout still holds.

Optional: ScriptRunner

type ScriptRunner interface {
    Eval(ctx context.Context, script string, keys []string, args ...any) (any, error)
}

App.EraseUserData removes a user's jobs from the queue through the store eraser NewRedisQueue registers, and that erasure runs as one Lua script the queue owns. The adapter passes it through; with go-redis:

func (c adapter) Eval(ctx context.Context, script string, keys []string, args ...any) (any, error) {
    return c.rdb.Eval(ctx, script, keys, args...).Result()
}

Redis runs a script with no other command in between. A sweep that reads one key at a time cannot prove a job gone, since a worker's Dequeue or Nack can move it into each key just after the sweep read that key, and a hash field read in one round trip and deleted in the next can belong to another user's claim by then. So an adapter without Eval keeps running the queue, but erasing a user's data from it fails with an error naming queue.ScriptRunner and removes nothing.

With Eval, the queue's own moves between its keys run as scripts too, each writing only while the source still holds the entry: Dequeue's claim from its pop record into the processing hash, its dead-lettering and requeues, Nack and Reclaim without CompareAndMover, Replay, and the promotion of due delayed jobs. A worker that read a job just before an erasure removed it would otherwise write the job back. With the script it finds the source empty and drops the job: Dequeue moves on to the next one, and Nack or Replay does nothing. The claim needs PopRecorder as well, since the pop record is what it checks; without one, a job Dequeue has popped exists only in the worker's memory until the claim writes it.

The script reads every entry of the main, delayed and dead lists and the pop and processing hashes, and the server serves no other client while it runs, so it takes longer the more jobs the queue holds. It matches a job by the user_id key json.Marshal writes at the top of the job, compared byte for byte with json.Marshal(userID), and never decodes the payload: Redis's cjson refuses JSON nested past 1,000 levels, which Enqueue accepts, and a job it could not decode would have been kept. A user id that is not valid UTF-8 is refused with an error, because json.Marshal stores every such id as "�" and the script cannot tell those users' jobs apart. It names all five keys, so on Redis Cluster use the same hash-tagged queue name as for CompareAndMover. Return an integer reply as an integer type (go-redis returns int64).

RPop must report an empty list as queue.ErrRedisEmpty. Drivers signal "nothing there" with their own sentinel (go-redis returns redis.Nil); translate it. An adapter that passes its driver's sentinel through unchanged turns every empty poll into a backend error, and a worker loop that only branches on ErrNoJob will fall through and handle a zero-valued Job.

Sentinel errors

queue.ErrNoJob       // Dequeue: nothing ready right now
queue.ErrQueueClosed // Enqueue: queue was already closed
queue.ErrDuplicateJob // Enqueue, Replay: a live job of the same type holds the UniqueKey
queue.ErrUniqueUnsupported // Enqueue: the backend cannot hold a UniqueKey (RedisQueue)
queue.ErrRedisEmpty  // RedisClient.RPop: the list is empty (adapters MUST
                     // map their driver's nil-sentinel onto this)

Common mistakes

• Not calling q.Start(ctx, interval) on RedisQueue. Without it, crashed-worker jobs strand in the processing hash indefinitely.
• Closing MemoryQueue before workers drain. Close waits for in-flight handlers to finish; call it after all producers are done.
• Replaying a job that is still pending. Replay only touches terminal (failed) entries; replaying a pending job is a no-op.
• Enqueuing personal data with no UserID. The payload column is exported verbatim and, without Job.UserID, no erasure can reach it.
• Acking or Nacking a hand-built Job. Both backends fence completions on the claim identity Dequeue minted (Job.ClaimToken on Redis, the claim_token column on the DB). Pass back the job Dequeue returned, not a fresh struct carrying only the ID, or the completion matches nothing and silently no-ops.
• Running the in-memory Scheduler on every replica. Multiple replicas can fire the same tick. Pin that mode to one process, or use DurableScheduler with DBQueue for fenced, transactionally deduplicated occurrences.
• Ignoring Nack errors. A Nack failure means the job stays in the processing hash (Redis) or claimed state (DB) and will be auto-reclaimed later, but log the error so you can spot connection issues early.