queue

Documentation

Overview

Package queue holds scrape jobs: a URL and an optional payload, leased to one worker at a time, retried on failure, and moved to a dead-letter state after too many attempts.

Memory keeps jobs in the process; Postgres keeps them in a table and lets many workers on many machines share it, using FOR UPDATE SKIP LOCKED so no two workers lease the same job. Both implement Queue and behave the same way.

A lease is identified by the job's attempt count. If a worker stalls past its lease, the job is handed to another worker with a higher attempt count, and the stalled worker's Complete or Fail is refused with ErrLeaseLost instead of overwriting the new worker's result.

Index

• Variables
• type DB
• type Job
• type Memory
  • func NewMemory(opts Options) *Memory
  • func (m *Memory) Complete(_ context.Context, j *Job) error
  • func (m *Memory) DeadLetters(_ context.Context, limit int) ([]Job, error)
  • func (m *Memory) Dequeue(_ context.Context) (*Job, error)
  • func (m *Memory) Enqueue(_ context.Context, j Job) (int64, error)
  • func (m *Memory) Fail(_ context.Context, j *Job, cause error, delay time.Duration) (bool, error)
  • func (m *Memory) Stats(_ context.Context) (Stats, error)
• type Options
• type Postgres
  • func NewPostgres(db DB, opts PostgresOptions) (*Postgres, error)
  • func (q *Postgres) Complete(ctx context.Context, j *Job) error
  • func (q *Postgres) DeadLetters(ctx context.Context, limit int) ([]Job, error)
  • func (q *Postgres) Dequeue(ctx context.Context) (*Job, error)
  • func (q *Postgres) Enqueue(ctx context.Context, j Job) (int64, error)
  • func (q *Postgres) Fail(ctx context.Context, j *Job, cause error, delay time.Duration) (bool, error)
  • func (q *Postgres) Migrate(ctx context.Context) error
  • func (q *Postgres) Stats(ctx context.Context) (Stats, error)
• type PostgresOptions
• type Queue
• type State
• type Stats

Variables

var (
	ErrEmpty     = errors.New("queue: no job is ready")
	ErrLeaseLost = errors.New("queue: job is not leased to this caller")
	ErrNoURL     = errors.New("queue: job has no URL")
)

Errors returned by both implementations.

Types

type DB

type DB interface {
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

DB is the part of pgx the Postgres queue uses. *pgxpool.Pool, *pgx.Conn and pgx.Tx all satisfy it.

type Job

type Job struct {
	ID          int64
	URL         string
	Payload     []byte
	State       State
	Attempts    int // leases so far, including the current one
	MaxAttempts int
	LastError   string
	RunAt       time.Time // not leased before this time
	CreatedAt   time.Time
}

Job is one unit of work.

type Memory

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

Memory is an in-process Queue. Jobs are lost when the process exits.

func NewMemory

func NewMemory(opts Options) *Memory

NewMemory returns an empty in-memory queue.

func (*Memory) Complete

func (m *Memory) Complete(_ context.Context, j *Job) error

Complete implements Queue.

func (*Memory) DeadLetters

func (m *Memory) DeadLetters(_ context.Context, limit int) ([]Job, error)

DeadLetters implements Queue.

func (*Memory) Dequeue

func (m *Memory) Dequeue(_ context.Context) (*Job, error)

Dequeue implements Queue. It leases the ready job with the earliest RunAt, then the lowest ID.

func (*Memory) Enqueue

func (m *Memory) Enqueue(_ context.Context, j Job) (int64, error)

Enqueue implements Queue.

func (*Memory) Fail

func (m *Memory) Fail(_ context.Context, j *Job, cause error, delay time.Duration) (bool, error)

Fail implements Queue.

func (*Memory) Stats

func (m *Memory) Stats(_ context.Context) (Stats, error)

Stats implements Queue.

type Options

type Options struct {
	// MaxAttempts is the default for jobs that set none. Default 5.
	MaxAttempts int
	// Lease is how long a dequeued job stays leased before another worker
	// may take it. Default 5 minutes.
	Lease time.Duration
}

Options tune a queue. The zero value is usable.

type Postgres

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

Postgres is a Queue stored in a Postgres table. Time is the database's clock, so workers on machines with skewed clocks still agree on leases.

func NewPostgres

func NewPostgres(db DB, opts PostgresOptions) (*Postgres, error)

NewPostgres returns a queue over db. It does not touch the database; call Migrate once to create the table.

func (*Postgres) Complete

func (q *Postgres) Complete(ctx context.Context, j *Job) error

Complete implements Queue.

func (*Postgres) DeadLetters

func (q *Postgres) DeadLetters(ctx context.Context, limit int) ([]Job, error)

DeadLetters implements Queue.

func (*Postgres) Dequeue

func (q *Postgres) Dequeue(ctx context.Context) (*Job, error)

Dequeue implements Queue. A job whose lease expired on its last attempt is moved to dead on the way.

func (*Postgres) Enqueue

func (q *Postgres) Enqueue(ctx context.Context, j Job) (int64, error)

Enqueue implements Queue.

func (*Postgres) Fail

func (q *Postgres) Fail(ctx context.Context, j *Job, cause error, delay time.Duration) (bool, error)

Fail implements Queue.

func (*Postgres) Migrate

func (q *Postgres) Migrate(ctx context.Context) error

Migrate creates the table and its index if they do not exist.

func (*Postgres) Stats

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

Stats implements Queue.

type PostgresOptions

type PostgresOptions struct {
	Options
	// Table holds the jobs. Default "forage_jobs". Several queues can
	// share a database by using different tables.
	Table string
}

PostgresOptions tune a Postgres queue.

type Queue

type Queue interface {
	// Enqueue adds a job and returns its ID. Only URL is read from the
	// job, plus MaxAttempts (zero for the queue's default) and RunAt (zero
	// for now) and Payload.
	Enqueue(ctx context.Context, j Job) (int64, error)
	// Dequeue leases the next ready job, or returns ErrEmpty.
	Dequeue(ctx context.Context) (*Job, error)
	// Complete marks a leased job done.
	Complete(ctx context.Context, j *Job) error
	// Fail records a failed attempt. The job runs again after delay unless
	// it has used all its attempts, in which case it is dead and Fail
	// reports dead = true.
	Fail(ctx context.Context, j *Job, cause error, delay time.Duration) (dead bool, err error)
	// DeadLetters returns up to limit dead jobs, oldest first.
	DeadLetters(ctx context.Context, limit int) ([]Job, error)
	// Stats counts jobs by state.
	Stats(ctx context.Context) (Stats, error)
}

Queue is a job queue. Implementations are safe for concurrent use.

type State

type State string

State is where a job is in its life.

const (
	Pending State = "pending"
	Running State = "running"
	Done    State = "done"
	Dead    State = "dead"
)

The states.

type Stats

type Stats struct {
	Pending, Running, Done, Dead int
}

Stats counts jobs by state.