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 failuresq.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()

NewDBQueue creates the queue_jobs table and its index if they do not
exist. Pass WithTable("my_jobs") to use a custom table name.

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

<!-- gofastr:compile
import "encoding/json"
import "time"
-->

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}

Scheduled jobs (future ScheduledAt) are invisible to Dequeue until
the moment passes. This lets you implement delayed processing without a
separate scheduler.

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:

<!-- gofastr:compile
import "github.com/DonaldMurillo/gofastr/battery/queue"
import "time"
-->

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.

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).

Inspecting jobs (Browsable)

All three backends implement Browsable:

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

ListJobs accepts a status string ("pending", "failed", "" for
all) and a limit. Jobs are returned newest-first. Stats returns a
JobStats map (status → count).

MemoryQueue and RedisQueue can only enumerate their dead-letter store,
so only "failed" (or "") returns results. DBQueue can enumerate any
status.

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.

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

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.

You can also call Reclaim directly from your own ticker:

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

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.

Every(d) schedules fire on a fixed interval; 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. 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 only when a Cron spec is invalid —
Every schedules never error, so existing callers that ignore the
return value are unaffected. 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)    }}()

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) are added to an
existing scheduler_schedules table by an idempotent migration during
NewDurableScheduler — no manual schema work is required on upgrade.

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)}

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 — 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.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. 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 no longer pending or claimed; live work is
never pruned. 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) to cancel
the handler's context at the deadline. 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 and the worker goroutine is respawned, so a poison
message cannot drain the worker pool.

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.

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 nowqueue.ErrQueueClosed // Enqueue: queue was already closedqueue.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.
• 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.