Documentation
¶
Overview ¶
Package sqlqueue implements a database-backed Queue driver that survives process restarts and works across replicas. It plugs into the parent queue package via the Queue interface.
The driver uses a single `jobs` table; the schema is dialect-aware and ships its own migration via Setup(ctx). Pop selects the next available row with `... ORDER BY available_at LIMIT 1` and updates it to `reserved_at = now()` in a transaction so concurrent workers don't claim the same job.
Usage:
q, err := sqlqueue.New(conn) // uses the project's
if err != nil { return err } // database.Connection
if err := q.Setup(ctx); err != nil { ... } // creates the table
queue.Dispatch(ctx, q, SendEmail{...})
worker := queue.NewWorker(q)
go worker.Run(ctx)
Index ¶
- type Option
- type Queue
- func (q *Queue) Ack(ctx context.Context, jobID string) error
- func (q *Queue) Len() int
- func (q *Queue) Nack(ctx context.Context, jobID string, retryAfter time.Duration) error
- func (q *Queue) Pop(ctx context.Context, wait time.Duration) (queue.Job, error)
- func (q *Queue) Push(ctx context.Context, j queue.Job) error
- func (q *Queue) Setup(ctx context.Context) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*Queue)
Option customises a Queue.
func WithVisibilityTimeout ¶
WithVisibilityTimeout sets how long a reserved-but-not-acked job stays invisible. Default 15 minutes — pick a value comfortably larger than the longest handler runtime. If the timeout is shorter than a slow-but-alive handler, orphan recovery will re-deliver the job while it is still being processed (it now bumps attempts, so MaxRetry still bounds the loop, but the work runs twice).
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue is a database-backed queue. Safe for concurrent use across goroutines and processes.
func New ¶
func New(conn *database.Connection, opts ...Option) (*Queue, error)
New constructs a Queue. On SQLite it lowers the connection pool to a single open connection and raises busy_timeout so concurrent reservers serialise on the writer lock instead of erroring with "database is locked".
func (*Queue) Len ¶
Len returns the row count of the jobs table (cheap on SQLite, OK on other dialects under modest sizes; for production metrics prefer a time-bucketed query).
func (*Queue) Nack ¶
Nack returns the job to the queue, incrementing attempts. If retryAfter > 0 the job stays invisible until that time.
func (*Queue) Pop ¶
Pop selects the next available job within wait. Reserves it via transactional update so concurrent workers cannot pick it up.
Source Files