queue

Oct 18, 2026

Changes in this version

+ const DefaultFailedTable

+ const DefaultTable

+ const MigrationID

+ var ErrPermanent = errors.New("queue: permanent failure, will not be retried")

+ func Migrations(o DatabaseOptions) []schema.Migration

+ type Attempt struct

+ MaxRetries int

+ Number int

+ func AttemptFrom(ctx context.Context) (Attempt, bool)

+ func (a Attempt) Last() bool

type JobEvent

+ Final bool

+ MaxRetries int

Oct 17, 2026

Changes in this version

+ const MaxFreeInterruptions

+ type Clock interface

+ SetClock func(now func() time.Time)

type Database

+ func (d *Database) SetClock(now func() time.Time)

type Memory

+ func (d *Memory) SetClock(now func() time.Time)

type Queue

+ func (q *Queue) AddObserver(obs Observer)

+ func (q *Queue) DispatchKindTx(ctx context.Context, ex database.Executor, kind string, payload any, ...) error

+ func (q *Queue) SetClock(now func() time.Time)

type WorkOptions

+ StopTimeout time.Duration

Oct 17, 2026

Changes in this version

+ var ErrInlineDelay = errors.New(...)

Oct 17, 2026

Changes in this version

+ var ErrReservationLost = errors.New("queue: job reservation lost to another worker")

Oct 16, 2026

Changes in this version

+ type JobEvent struct

+ Attempt int

+ Duration time.Duration

+ Err error

+ ID string

+ Kind string

+ Queue string

+ Status JobStatus

+ type JobStatus string

+ const JobDispatched

+ const JobFailed

+ const JobProcessed

+ type Observer interface

+ ObserveJob func(ctx context.Context, e JobEvent)

type Queue

+ func (q *Queue) SetObserver(obs Observer)

Oct 16, 2026

Changes in this version

+ var ErrDuplicateKind = errors.New("queue: job kind already registered")

+ var ErrNoTx = errors.New("queue: driver does not support DispatchTx")

+ var ErrUnknownKind = errors.New("queue: no handler for job kind")

+ func Constant(d time.Duration) func(int) time.Duration

+ func Dispatched[J Job](f *Fake) []J

+ func Exponential(base time.Duration) func(int) time.Duration

+ type Database struct

+ func NewDatabase(db *database.DB, o DatabaseOptions) (*Database, error)

+ func (d *Database) Ack(ctx context.Context, m *Message) error

+ func (d *Database) CreateTables(ctx context.Context) error

+ func (d *Database) Fail(ctx context.Context, m *Message) error

+ func (d *Database) Failed(ctx context.Context) ([]FailedJob, error)

+ func (d *Database) Forget(ctx context.Context, ids ...string) (int, error)

+ func (d *Database) Pending(ctx context.Context) (int, error)

+ func (d *Database) Pop(ctx context.Context, queues []string) (*Message, error)

+ func (d *Database) Push(ctx context.Context, m *Message) error

+ func (d *Database) PushTx(ctx context.Context, ex database.Executor, m *Message) error

+ func (d *Database) Release(ctx context.Context, m *Message) error

+ func (d *Database) Retry(ctx context.Context, ids ...string) (int, error)

+ type DatabaseOptions struct

+ FailedTable string

+ Grace time.Duration

+ Now func() time.Time

+ Table string

+ type Driver interface

+ Ack func(ctx context.Context, m *Message) error

+ Fail func(ctx context.Context, m *Message) error

+ Pop func(ctx context.Context, queues []string) (*Message, error)

+ Push func(ctx context.Context, m *Message) error

+ Release func(ctx context.Context, m *Message) error

+ type FailedJob struct

+ FailedAt time.Time

+ type FailedStore interface

+ Failed func(ctx context.Context) ([]FailedJob, error)

+ Forget func(ctx context.Context, ids ...string) (int, error)

+ Retry func(ctx context.Context, ids ...string) (int, error)

+ type Fake struct

+ func NewFake() *Fake

+ func (f *Fake) Messages() []Message

+ func (f *Fake) Push(ctx context.Context, m *Message) error

+ type Inline interface

+ Inline func() bool

+ type Job interface

+ Kind func() string

+ type Memory struct

+ func NewMemory() *Memory

+ func (d *Memory) Ack(_ context.Context, m *Message) error

+ func (d *Memory) Fail(_ context.Context, m *Message) error

+ func (d *Memory) Failed() []Message

+ func (d *Memory) Pending() int

+ func (d *Memory) Pop(_ context.Context, queues []string) (*Message, error)

+ func (d *Memory) Push(_ context.Context, m *Message) error

+ func (d *Memory) Release(_ context.Context, m *Message) error

+ type Message struct

+ Attempts int

+ AvailableAt time.Time

+ CreatedAt time.Time

+ ID string

+ Kind string

+ LastError string

+ MaxRetries int

+ Payload json.RawMessage

+ Queue string

+ Timeout time.Duration

+ type Option func(*options)

+ func Backoff(fn func(attempt int) time.Duration) Option

+ func Delay(d time.Duration) Option

+ func OnQueue(name string) Option

+ func Retries(n int) Option

+ func Timeout(d time.Duration) Option

+ type Queue struct

+ func New(driver Driver, log *slog.Logger) *Queue

+ func (q *Queue) Dispatch(ctx context.Context, j Job, opts ...Option) error

+ func (q *Queue) DispatchKind(ctx context.Context, kind string, payload any, opts ...Option) error

+ func (q *Queue) DispatchTx(ctx context.Context, ex database.Executor, j Job, opts ...Option) error

+ func (q *Queue) Driver() Driver

+ func (q *Queue) Message(j Job, opts ...Option) (*Message, error)

+ func (q *Queue) Process(ctx context.Context, m *Message) error

+ func (q *Queue) Register(regs ...Registration) error

+ func (q *Queue) Work(ctx context.Context, o WorkOptions) error

+ type Registration struct

+ func HandleKind[T any](kind string, fn func(ctx context.Context, v T) error, opts ...Option) Registration

+ func Handle[J Job](fn func(ctx context.Context, j J) error, opts ...Option) Registration

+ func (r Registration) Kind() string

+ type Sync struct

+ func (Sync) Ack(context.Context, *Message) error

+ func (Sync) Fail(context.Context, *Message) error

+ func (Sync) Inline() bool

+ func (Sync) Pop(context.Context, []string) (*Message, error)

+ func (Sync) Push(context.Context, *Message) error

+ func (Sync) Release(context.Context, *Message) error

+ type TxPusher interface

+ PushTx func(ctx context.Context, ex database.Executor, m *Message) error

+ type WorkOptions struct

+ Concurrency int

+ MaxJobs int

+ Queues []string

+ Sleep time.Duration

+ StopWhenEmpty bool