queue

package
v0.37.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 20, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package queue is part of the GoFastr framework. See https://github.com/DonaldMurillo/gofastr for documentation.

Package queue provides a pluggable job queue with in-memory and Redis backends, a goroutine pool for concurrent processing, and scheduled job support.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrQueueClosed = errors.New("queue is closed")
	ErrNoJob       = errors.New("no job available")
)

Sentinel errors.

Functions

This section is empty.

Types

type Browsable 

type Browsable interface {
	// ListJobs returns up to limit jobs in the given status; pass an
	// empty status to return all jobs regardless of state. Jobs are
	// ordered newest-first by created_at.
	ListJobs(ctx context.Context, status string, limit int) ([]Job, error)
	// Stats returns counts grouped by status. Cheap by design — admin
	// dashboards may poll it.
	Stats(ctx context.Context) (JobStats, error)
}

Browsable is the optional read-only inspection interface — implemented by DBQueue so admin tooling can list and aggregate jobs without guessing at the underlying schema. Memory and Redis queues may implement it later; admin code that depends on it should type-assert.

type DBQueue 

type DBQueue struct {
	// contains filtered or unexported fields
}

DBQueue is a SQL-backed queue. Jobs persist in a single table; Dequeue claims a row atomically so multiple consumers can race safely.

Postgres uses FOR UPDATE SKIP LOCKED — the canonical pattern for queue fan-out without distributed locks. SQLite uses a SERIALIZABLE-friendly SELECT-then-UPDATE inside a tx; the table-level lock SQLite takes on BEGIN IMMEDIATE serialises writers naturally.

func NewDBQueue 

func NewDBQueue(db *sql.DB, opts ...DBQueueOption) (*DBQueue, error)

NewDBQueue constructs a DBQueue and ensures its backing table exists. Probes the dialect once via SELECT version(); falls back to SQLite. Panics if the table name contains unsafe characters.

func (*DBQueue) Ack 

func (q *DBQueue) Ack(ctx context.Context, jobID string) error

Ack permanently removes the job — work is done, no replay needed.

func (*DBQueue) Close 

func (q *DBQueue) Close() error

Close stops worker goroutines started by Start. Idempotent.

func (*DBQueue) Dequeue 

func (q *DBQueue) Dequeue(ctx context.Context, types ...string) (Job, error)

Dequeue claims the highest-priority eligible job in a single atomic step. Returns ErrNoJob when nothing is ready (no pending row whose scheduled_at has passed). Claims from ANY lane — dedicated lane workers use the internal dequeue with a lane filter instead.

func (*DBQueue) Enqueue 

func (q *DBQueue) Enqueue(ctx context.Context, job Job) error

Enqueue inserts a job. Fills in ID/CreatedAt/MaxAttempts/ScheduledAt defaults when zero-valued so callers can pass {Type, Payload} only.

func (*DBQueue) ListJobs 

func (q *DBQueue) ListJobs(ctx context.Context, status string, limit int) ([]Job, error)

ListJobs implements Browsable. Returns up to limit jobs in the supplied status, newest-first. Empty status returns all jobs regardless of state. limit <= 0 defaults to 100.

func (*DBQueue) Nack 

func (q *DBQueue) Nack(ctx context.Context, jobID string) error

Nack returns a claimed job to the queue (status=pending) when it still has retry attempts left; otherwise marks it 'failed' for later inspection. When backoff is enabled (see WithBackoff), a requeued job's scheduled_at is pushed into the future so a flapping handler can't burn through every attempt in a tight loop.

func (*DBQueue) RegisterHandler 

func (q *DBQueue) RegisterHandler(jobType string, h Handler)

RegisterHandler binds a job type to a handler. Safe to call concurrently with a running worker loop.

func (*DBQueue) Replay 

func (q *DBQueue) Replay(ctx context.Context, jobID string) error

Replay implements Replayable: it resets a terminally-failed job to pending so a worker picks it up again — attempts cleared, scheduled immediately. The `AND status='failed'` clause makes it idempotent and safe: replaying an unknown, pending, running, or claimed job matches no row and is a no-op, so it can never double-run an in-flight job or resurrect a non-terminal one.

func (*DBQueue) SetGate added in v0.17.0 

func (q *DBQueue) SetGate(gate func(jobType string) bool)

SetGate installs a gate checked in the worker loop after a handler is resolved. When gate returns false the job is released back to pending (without consuming a retry attempt) and rescheduled slightly into the future to avoid a hot loop. Framework code uses it to defer jobs owned by a disabled module. Pass nil to clear.

func (*DBQueue) SetLeaseTimeout 

func (q *DBQueue) SetLeaseTimeout(d time.Duration)

SetLeaseTimeout adjusts the in-flight lease duration after construction. See WithLeaseTimeout for semantics. Safe to call concurrently with a running worker loop.

func (*DBQueue) Start 

func (q *DBQueue) Start(ctx context.Context)

Start launches the shared worker pool (q.workers goroutines that claim any lane) plus one goroutine per dedicated lane worker added via WithLaneWorkers. Each loops dequeue → handle → Ack/Nack until Close. A worker goroutine that dies for any reason (including a panic escaping the handler-recover guard) is respawned so the pool can never be permanently drained by a poison message.

func (*DBQueue) Stats 

func (q *DBQueue) Stats(ctx context.Context) (JobStats, error)

Stats implements Browsable. Aggregates per-status counts over the whole table. Cheap: a single GROUP BY scan.

type DBQueueOption 

type DBQueueOption func(*DBQueue)

DBQueueOption configures DBQueue construction.

func WithBackoff 

func WithBackoff(base, max time.Duration) DBQueueOption

WithBackoff enables exponential retry backoff. On a Nack with retries remaining, scheduled_at is advanced by base*2^(attempts-1) — so the first retry waits ~base, the second ~2*base, and so on — capped at max. A non-positive base disables backoff (jobs retry immediately, the default). A non-positive max means uncapped. Mirrors the webhook battery's retry backoff so the two batteries behave consistently.

func WithDBHandlerTimeout added in v0.11.0 

func WithDBHandlerTimeout(d time.Duration) DBQueueOption

WithDBHandlerTimeout caps a single handler invocation's wall-clock budget. The job's context is cancelled at the deadline, so a black-holed dependency (an SMTP host that never responds, a hung HTTP call) can't wedge a worker forever — critical with the default single worker, where one stuck job stalls the entire queue. Zero (default) means no timeout; set it whenever handlers touch the network. (Named distinctly from the MemoryQueue's WithHandlerTimeout because both live in this package.)

func WithDBLaneWorkers added in v0.19.0 

func WithDBLaneWorkers(lane string, n int) DBQueueOption

WithDBLaneWorkers adds n dedicated worker goroutines that ONLY claim jobs whose Lane equals the given lane, on top of the shared workers from WithWorkers. This is the mechanism that prevents bulk backfills from starving urgent jobs: even when every shared worker is busy running a long-running bulk handler, the dedicated lane workers keep draining the reserved lane. Multiple calls for different lanes each add their own workers; multiple calls for the same lane sum. Priority ordering still selects among pending jobs within a worker's claim set. Panics if n <= 0 or lane is "" (use WithWorkers for the shared/default pool). Named with the DB prefix to match WithDBHandlerTimeout / WithDBLogger (both backends share this package, so the MemoryQueue variant is WithLaneWorkers).

func WithDBLogger added in v0.18.0 

func WithDBLogger(l *slog.Logger) DBQueueOption

WithDBLogger sets the logger used for handler-failure (WARN) and dead-letter (ERROR) records emitted from the worker loop. Defaults to slog.Default(); passing nil restores the default. Without it a failing handler dead-letters silently — you lose the job and the reason.

func WithLeaseTimeout 

func WithLeaseTimeout(d time.Duration) DBQueueOption

WithLeaseTimeout sets how long a claimed-but-unacked job may stay in-flight before it is considered abandoned (the worker crashed/was killed) and becomes eligible for re-dequeue. Defaults to 5 minutes.

func WithTable 

func WithTable(name string) DBQueueOption

WithTable overrides the default "queue_jobs" table name.

func WithWorkers 

func WithWorkers(n int) DBQueueOption

WithWorkers sets the number of background worker goroutines started by Start(). Defaults to 1 when not set.

type DurableScheduleBuilder added in v0.33.0 

type DurableScheduleBuilder struct {
	// contains filtered or unexported fields
}

DurableScheduleBuilder configures one persisted recurring schedule.

func (*DurableScheduleBuilder) Job added in v0.33.0 

func (b *DurableScheduleBuilder) Job(jobType string, payload any) *DurableScheduleBuilder

Job sets the queue job type and JSON payload for a durable schedule.

func (*DurableScheduleBuilder) Lane added in v0.37.0 

func (b *DurableScheduleBuilder) Lane(name string) *DurableScheduleBuilder

Lane routes every Job fired by this schedule to the given capacity- reservation lane (see DBQueue.WithDBLaneWorkers). Empty (the default) lands the job on the shared/default lane. The value persists alongside the schedule definition; re-registering the same schedule ID updates it.

func (*DurableScheduleBuilder) MaxAttempts added in v0.37.0 

func (b *DurableScheduleBuilder) MaxAttempts(n int) *DurableScheduleBuilder

MaxAttempts bounds the retry ceiling carried into every Job fired by this schedule. Zero (the default) lets Enqueue resolve it to 3 — matching the pre-options behaviour. The value persists alongside the schedule definition; re-registering the same schedule ID updates it.

func (*DurableScheduleBuilder) Priority added in v0.37.0 

func (b *DurableScheduleBuilder) Priority(p int) *DurableScheduleBuilder

Priority sets the priority carried into every Job fired by this schedule. Higher integers are dequeued first. Defaults to 0. The value persists alongside the schedule definition; re-registering the same schedule ID updates it.

func (*DurableScheduleBuilder) Register added in v0.33.0 

func (b *DurableScheduleBuilder) Register() error

Register persists the schedule definition without resetting an existing watermark.

func (*DurableScheduleBuilder) RegisterAt added in v0.33.0 

func (b *DurableScheduleBuilder) RegisterAt(base time.Time) error

RegisterAt is Register with a deterministic first-run anchor.

type DurableScheduler added in v0.33.0 

type DurableScheduler struct {
	// contains filtered or unexported fields
}

DurableScheduler persists schedule watermarks and occurrences in the same SQL database as a DBQueue. A heartbeat-expiry lease avoids redundant evaluation, while unique occurrences and transactionally enqueued jobs are the deduplication authority.

func NewDurableScheduler added in v0.33.0 

func NewDurableScheduler(q *DBQueue, cfg DurableSchedulerConfig) (*DurableScheduler, error)

NewDurableScheduler creates a replica-safe scheduler backed by q's SQL database. It creates the schedule, occurrence, and lease tables when absent.

func (*DurableScheduler) Cron added in v0.33.0 

func (s *DurableScheduler) Cron(id, spec string) *DurableScheduleBuilder

Cron defines a durable cron schedule. id is the stable identity used to preserve its watermark across restarts and replica registration.

func (*DurableScheduler) Every added in v0.33.0 

func (s *DurableScheduler) Every(id string, interval time.Duration) *DurableScheduleBuilder

Every defines a durable fixed-interval schedule. id is the stable identity used to preserve its watermark across restarts and replica registration.

func (*DurableScheduler) RunOnce added in v0.33.0 

func (s *DurableScheduler) RunOnce(ctx context.Context, now time.Time) error

RunOnce evaluates every due schedule at now. A replica that does not own the current fenced lease returns nil without side effects.

func (*DurableScheduler) Start added in v0.33.0 

func (s *DurableScheduler) Start(ctx context.Context) error

Start runs durable evaluation until ctx is cancelled. Registration wakes the loop immediately. The lease heartbeat is the maximum sleep; the earliest persisted next-run timestamp wakes it sooner.

type DurableSchedulerConfig added in v0.33.0 

type DurableSchedulerConfig struct {
	OwnerID               string
	LeaseDuration         time.Duration
	OccurrenceRetention   time.Duration
	MaxCatchUpOccurrences int
}

DurableSchedulerConfig identifies one scheduler replica and controls how quickly another replica may reclaim its leadership lease after a heartbeat stops. OwnerID defaults to a random process-local ID. LeaseDuration defaults to 30 seconds. OccurrenceRetention defaults to 30 days; a negative value disables occurrence pruning. MaxCatchUpOccurrences defaults to 1000 and bounds the history rows materialized after downtime.

type Handler 

type Handler func(ctx context.Context, job Job) error

Handler processes a job. Return a non-nil error to trigger a retry.

type Job 

type Job struct {
	ID           string          `json:"id"`
	OccurrenceID string          `json:"occurrence_id,omitempty"`
	Type         string          `json:"type"`
	Payload      json.RawMessage `json:"payload"`
	Priority     int             `json:"priority"`
	Lane         string          `json:"lane"`
	Attempts     int             `json:"attempts"`
	MaxAttempts  int             `json:"max_attempts"`
	CreatedAt    time.Time       `json:"created_at"`
	ScheduledAt  time.Time       `json:"scheduled_at"`
}

Job represents a unit of work enqueued for asynchronous processing.

type JobStats 

type JobStats map[string]int

JobStats is a snapshot of job counts grouped by status. The keys are status names ("pending", "running", "failed", "dead").

type MemoryQueue 

type MemoryQueue struct {
	// contains filtered or unexported fields
}

MemoryQueue is an in-memory queue backed by a goroutine pool. Pending jobs live in a priority heap (Priority DESC, enqueue-order ASC tiebreak) so higher-priority jobs are always taken first — priority is honoured, not ignored.

func NewMemoryQueue 

func NewMemoryQueue(workers int, opts ...MemoryQueueOption) *MemoryQueue

NewMemoryQueue creates a new in-memory queue with the given number of shared workers. Optional functional options (e.g. WithHandlerTimeout, WithLaneWorkers) may be passed. Shared workers take jobs from any lane by priority; dedicated lane workers (added via WithLaneWorkers) take only their own lane.

func (*MemoryQueue) Ack 

func (q *MemoryQueue) Ack(_ context.Context, jobID string) error

Ack confirms a manually-dequeued job is done, discarding any tracked in-flight copy. For jobs processed by the automatic worker pool it is a no-op (those are auto-acknowledged after successful handler execution).

func (*MemoryQueue) Close 

func (q *MemoryQueue) Close() error

Close drains pending jobs and waits for all workers to finish. It signals the cond so every waiting worker wakes, processes any remaining claimable job, then exits when the store is empty. Shared workers drain every lane; lane workers drain only their own. No goroutine leaks, no panics.

func (*MemoryQueue) Dequeue 

func (q *MemoryQueue) Dequeue(ctx context.Context, types ...string) (Job, error)

Dequeue retrieves the highest-priority job from the pending store, optionally filtered by type. Useful for manual consumption without the automatic worker pool. It is non-blocking: returns ErrNoJob when nothing matches. Jobs of every lane are eligible (dedicated lane filtering is a worker-pool concept, not a manual-consumption one).

func (*MemoryQueue) Enqueue 

func (q *MemoryQueue) Enqueue(ctx context.Context, job Job) error

Enqueue adds a job to the priority-ordered pending store. If the job has no ID, one is generated. The store is unbounded, so Enqueue never blocks on capacity (unlike the old buffered channel); it returns ErrQueueClosed only when the queue has been closed.

func (*MemoryQueue) ListJobs 

func (q *MemoryQueue) ListJobs(_ context.Context, status string, limit int) ([]Job, error)

ListJobs implements Browsable for the in-memory backend. The only state it can enumerate is the retained dead-letter set, so it returns those jobs for status "failed" (or an empty/"all" status), newest-first, and nothing for any other status — pending/claimed jobs live transiently on the pending heap. limit <= 0 defaults to 100.

func (*MemoryQueue) Nack 

func (q *MemoryQueue) Nack(ctx context.Context, jobID string) error

Nack marks a manually-dequeued job as failed: it increments the attempt counter and re-enqueues the job when retries remain, otherwise drops it. The job must have been handed out by Dequeue (the in-flight set is consulted by ID). Jobs processed by the automatic worker pool retry internally and are never in the in-flight set; calling Nack for one is a harmless no-op.

func (*MemoryQueue) RegisterHandler 

func (q *MemoryQueue) RegisterHandler(jobType string, handler Handler)

RegisterHandler registers a handler function for a given job type.

func (*MemoryQueue) Replay 

func (q *MemoryQueue) Replay(ctx context.Context, jobID string) error

Replay implements Replayable: it moves a retained terminally-failed job back onto the pending set with Attempts reset to 0 so Dequeue returns it again. Idempotent and safe: replaying an unknown or non-failed id matches no retained job and is a no-op (nil), never a double-enqueue.

func (*MemoryQueue) SetGate added in v0.17.0 

func (q *MemoryQueue) SetGate(gate func(jobType string) bool)

SetGate installs a gate checked in processJob after handlerFor. When gate returns false the job is re-enqueued with a short delay so it runs when the module re-enables. Framework code uses it to defer jobs owned by a disabled module. Pass nil to clear.

func (*MemoryQueue) Start 

func (q *MemoryQueue) Start()

Start launches the shared worker goroutines plus one goroutine per dedicated lane worker (WithLaneWorkers). Must be called before enqueuing jobs if you want automatic processing.

func (*MemoryQueue) Stats 

func (q *MemoryQueue) Stats(_ context.Context) (JobStats, error)

Stats implements Browsable. The in-memory backend can only count the retained dead-letter jobs (pending/claimed jobs are transient on the heap), so it reports those under "failed".

type MemoryQueueOption added in v0.3.1 

type MemoryQueueOption func(*MemoryQueue)

MemoryQueueOption is a functional option for NewMemoryQueue.

func WithHandlerTimeout added in v0.3.1 

func WithHandlerTimeout(d time.Duration) MemoryQueueOption

WithHandlerTimeout sets the per-job execution timeout for the automatic worker pool. Jobs that run longer than the timeout have their context cancelled, which the handler should respect (the job is then retried or dead-lettered as usual). Defaults to 30 s.

func WithLaneWorkers added in v0.19.0 

func WithLaneWorkers(lane string, n int) MemoryQueueOption

WithLaneWorkers adds n dedicated worker goroutines that ONLY take jobs whose Lane equals the given lane, on top of the shared workers passed to NewMemoryQueue. This mirrors DBQueue.WithLaneWorkers and prevents a bulk backfill from starving urgent jobs even under worker saturation. Multiple calls for different lanes each add their own workers; multiple calls for the same lane sum. Panics if n <= 0 or lane is "" (use the shared worker count for the default lane).

func WithLogger added in v0.18.0 

func WithLogger(l *slog.Logger) MemoryQueueOption

WithLogger sets the logger used for handler-failure (WARN) and dead-letter (ERROR) records emitted from the worker pool. Defaults to slog.Default(); passing nil restores the default. Unprefixed to match the MemoryQueue's WithHandlerTimeout (the DBQueue uses WithDBLogger).

type Queue 

type Queue interface {
	// Enqueue adds a job to the queue.
	Enqueue(ctx context.Context, job Job) error
	// Dequeue retrieves and removes the next available job, optionally filtered by type.
	Dequeue(ctx context.Context, types ...string) (Job, error)
	// Ack confirms successful processing of a job.
	Ack(ctx context.Context, jobID string) error
	// Nack marks a job as failed and triggers retry logic.
	Nack(ctx context.Context, jobID string) error
	// Close gracefully shuts down the queue, draining in-progress work.
	Close() error
}

Queue is the interface that every queue backend must implement.

type RedisClient 

type RedisClient interface {
	LPush(ctx context.Context, key string, values ...interface{}) error
	RPop(ctx context.Context, key string) (string, error)
	HSet(ctx context.Context, key string, values ...interface{}) error
	HGet(ctx context.Context, key, field string) (string, error)
	// HGetAll returns every field→value pair in the hash. Used by Reclaim to
	// scan the processing set for expired in-flight jobs.
	HGetAll(ctx context.Context, key string) (map[string]string, error)
	HDel(ctx context.Context, key string, fields ...string) error
	Del(ctx context.Context, keys ...string) error
	// LRange returns the elements of the list at key in the inclusive range
	// [start, stop]; negative indices count from the tail (-1 is the last
	// element). Used by Replay to read the dead-letter list.
	LRange(ctx context.Context, key string, start, stop int64) ([]string, error)
	// LRem removes up to count occurrences of value from the list at key and
	// returns the number removed. Used by Replay to pull one entry off the
	// dead-letter list.
	LRem(ctx context.Context, key string, count int64, value interface{}) (int64, error)
}

RedisClient defines the minimal Redis operations needed by RedisQueue. This is an interface so callers can inject any Redis client (go-redis, redigo, etc.) without this package importing a specific driver.

type RedisQueue 

type RedisQueue struct {
	// contains filtered or unexported fields
}

RedisQueue implements the Queue interface backed by Redis lists and hashes. It supports a visibility timeout for in-flight jobs and a dead letter queue for jobs that exceed MaxAttempts.

func NewRedisQueue 

func NewRedisQueue(client RedisClient, queueName string) *RedisQueue

NewRedisQueue creates a new Redis-backed queue.

func (*RedisQueue) Ack 

func (q *RedisQueue) Ack(ctx context.Context, jobID string) error

Ack removes a single job from the processing queue after successful handling.

func (*RedisQueue) Close 

func (q *RedisQueue) Close() error

Close is a no-op for RedisQueue — the caller manages the Redis connection.

func (*RedisQueue) Dequeue 

func (q *RedisQueue) Dequeue(ctx context.Context, types ...string) (Job, error)

Dequeue pops a job from the Redis list and moves it to the processing queue. If types are specified, only jobs matching one of those types are returned; non-matching jobs are pushed back onto the list.

func (*RedisQueue) Enqueue 

func (q *RedisQueue) Enqueue(ctx context.Context, job Job) error

Enqueue pushes a job onto the Redis list, applying defaults for ID, CreatedAt, and MaxAttempts when not set.

func (*RedisQueue) ListJobs added in v0.3.1 

func (q *RedisQueue) ListJobs(ctx context.Context, status string, limit int) ([]Job, error)

ListJobs implements Browsable for the Redis backend. The only durable job state accessible without a scan of the full main/processing lists is the dead-letter queue, so this returns dead jobs for status "failed" (or an empty/"all" status) and nothing for any other status value. Jobs are returned newest-first (head of the Redis list) up to limit entries. limit <= 0 defaults to 100.

func (*RedisQueue) Nack 

func (q *RedisQueue) Nack(ctx context.Context, jobID string) error

Nack handles a failed job. If retries remain, it re-enqueues the job; otherwise it moves it to the dead letter queue.

func (*RedisQueue) Reclaim 

func (q *RedisQueue) Reclaim(ctx context.Context) (int, error)

Reclaim scans the processing set for in-flight jobs whose visibility timeout has passed (the worker that claimed them crashed before Ack/Nack), re-enqueues them onto the main list, and removes the stale processing entry. Returns the number of jobs re-delivered. Call it periodically (e.g. from a background ticker) to make in-flight Redis work crash-safe.

func (*RedisQueue) Replay 

func (q *RedisQueue) Replay(ctx context.Context, jobID string) error

Replay implements Replayable: it pulls a terminally-failed job off the dead-letter list and re-enqueues it onto the main queue with its attempts counter reset, so it gets a full set of retries again. It is idempotent — replaying an unknown job ID is a no-op (returns nil), matching DBQueue.Replay.

The entry is LPush'd back onto the main queue first and only removed from the dead list on success, so a failure between the two ops leaves the job on the dead list (recoverable) rather than dropping it. A crash in that window can leave one copy on each list; the next Replay/Dequeue tolerates the duplicate.

func (*RedisQueue) SetVisibilityTimeout 

func (q *RedisQueue) SetVisibilityTimeout(d time.Duration)

SetVisibilityTimeout configures how long a job can be in-flight before it is considered abandoned and eligible for re-delivery.

func (*RedisQueue) Start added in v0.3.2 

func (q *RedisQueue) Start(ctx context.Context, interval time.Duration)

Start launches a background goroutine that calls Reclaim on every tick to re-enqueue in-flight jobs whose visibility timeout has expired (e.g. because the worker that claimed them crashed before Ack/Nack). It mirrors DBQueue's built-in lease-expiry reclaim so crashed-worker jobs are not silently stranded.

The goroutine exits when ctx is cancelled. interval controls how often Reclaim is called; a value <= 0 defaults to 30 seconds. Typical use: 

q.Start(ctx, 30*time.Second)

func (*RedisQueue) Stats added in v0.3.1 

func (q *RedisQueue) Stats(ctx context.Context) (JobStats, error)

Stats implements Browsable for the Redis backend. It reports the count of dead-lettered jobs under the "failed" key; pending/in-flight jobs are not enumerable without a full scan and are omitted. Cheap: a single LRange(0, -1) length read.

type Replayable 

type Replayable interface {
	// Replay resets a terminally-failed job back to pending so it is picked up
	// again (attempts counter cleared, scheduled immediately). It MUST only
	// touch terminal ('failed') rows and be idempotent: replaying an unknown,
	// pending, or running job is a no-op, not an error or a double-run.
	Replay(ctx context.Context, jobID string) error
}

Replayable is the optional capability for re-queuing a dead-lettered job — implemented by DBQueue (the durable backend). Admin tooling type-asserts for it and only offers a "replay" action when the backend supports it. Memory and Redis queues don't implement it yet (memory drops dead jobs; redis's dead-list isn't readable through RedisClient).

type ScheduleBuilder 

type ScheduleBuilder struct {
	// contains filtered or unexported fields
}

ScheduleBuilder provides a fluent API for building scheduled jobs.

func (*ScheduleBuilder) Job 

func (b *ScheduleBuilder) Job(jobType string, payload any) *ScheduleBuilder

Job sets the job type and payload for the scheduled job.

func (*ScheduleBuilder) Lane added in v0.37.0 

func (b *ScheduleBuilder) Lane(name string) *ScheduleBuilder

Lane routes every Job fired by this schedule to the given capacity- reservation lane (see DBQueue.WithDBLaneWorkers / MemoryQueue.WithLaneWorkers). Empty (the default) lands the job on the shared/default lane.

func (*ScheduleBuilder) MaxAttempts added in v0.37.0 

func (b *ScheduleBuilder) MaxAttempts(n int) *ScheduleBuilder

MaxAttempts bounds the retry ceiling carried into every Job fired by this schedule. Zero (the default) lets Enqueue resolve it to 3 — matching the pre-options behaviour.

func (*ScheduleBuilder) Priority added in v0.37.0 

func (b *ScheduleBuilder) Priority(p int) *ScheduleBuilder

Priority sets the priority carried into every Job fired by this schedule. Higher integers are dequeued first. Defaults to 0.

func (*ScheduleBuilder) Register 

func (b *ScheduleBuilder) Register() error

Register adds the scheduled job to the scheduler, computing the first NextRun relative to the current wall clock. It returns an error only when a Cron schedule's spec is invalid; interval (Every) schedules never error, so existing callers that ignore the return value are unaffected.

func (*ScheduleBuilder) RegisterAt added in v0.3.3 

func (b *ScheduleBuilder) RegisterAt(base time.Time) error

RegisterAt is like Register but anchors the first NextRun to base instead of the wall clock. It exists so cron schedules can be registered deterministically (tests, replayed fixtures) without depending on time.Now().

type ScheduledJob 

type ScheduledJob struct {
	Type     string          `json:"type"`
	Payload  json.RawMessage `json:"payload"`
	Interval time.Duration   `json:"interval"`
	NextRun  time.Time       `json:"next_run"`

	// Lane, Priority, and MaxAttempts are carried into every Job fired by
	// this schedule. They default to "", 0, and 0 (which enqueueWith
	// resolves to 3) — matching the pre-options behaviour for callers that
	// never set them.
	Lane        string `json:"lane,omitempty"`
	Priority    int    `json:"priority"`
	MaxAttempts int    `json:"max_attempts"`
	// contains filtered or unexported fields
}

ScheduledJob defines a recurring job configuration.

A schedule fires on either a fixed Interval (set via Scheduler.Every) or a cron expression (set via Scheduler.Cron). Interval and cron are mutually exclusive: when cron is non-nil the Interval field is unused and NextRun is advanced by the cron expression instead.

type Scheduler 

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler is the non-durable, single-process scheduler. Its watermarks live only in memory. Use DurableScheduler with DBQueue when multiple replicas or restart continuity must be safe.

func NewInMemoryScheduler added in v0.33.0 

func NewInMemoryScheduler(queues ...Queue) *Scheduler

NewInMemoryScheduler explicitly creates the non-durable, single-process scheduler. It is equivalent to NewScheduler.

func NewScheduler 

func NewScheduler(queues ...Queue) *Scheduler

NewScheduler creates the non-durable, single-process Scheduler that dispatches to the given queues. Enqueue errors are logged via slog.Default(). NewInMemoryScheduler is the explicit spelling for new code.

func NewSchedulerWithLogger added in v0.3.1 

func NewSchedulerWithLogger(q Queue, logger *slog.Logger) *Scheduler

NewSchedulerWithLogger creates a new Scheduler with an explicit logger. Pass a non-nil *slog.Logger to control where enqueue-error messages are routed; passing nil falls back to slog.Default().

func (*Scheduler) Cron added in v0.3.3 

func (s *Scheduler) Cron(spec string) *ScheduleBuilder

Cron returns a ScheduleBuilder that fires when the cron expression's next time arrives. The spec accepts the standard 5-field syntax plus the @shortcuts (e.g. "0 2 * * *" for every day at 02:00, or "@daily"); it is parsed by framework/cron.Parse, so the queue does not carry a second cron parser. Spec errors surface from Register / RegisterAt, not here, so the fluent chain stays clean.

func (*Scheduler) Every 

func (s *Scheduler) Every(interval time.Duration) *ScheduleBuilder

Every returns a ScheduleBuilder that fires on a fixed interval.

func (*Scheduler) Start 

func (s *Scheduler) Start(ctx context.Context)

Start begins the scheduling loop. It blocks until ctx is cancelled.

The loop re-reads the schedule set (under lock) on every tick, so jobs registered AFTER Start still fire — the natural wiring is "start subsystems, then register jobs", and an empty-at-Start scheduler that exited would silently drop everything registered later. The tick cadence adapts to the finest live interval each pass, so adding a sub-minute schedule after Start takes effect on the next tick.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.