liteq

package module
v0.0.3-alpha Latest Latest
Warning

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

 Go to latest Published: Apr 5, 2026 License: MIT Imports: 24 Imported by: 0

README

liteq

CI

A PostgreSQL-backed task queue for Go that eliminates extra infrastructure.

If your app already runs on Postgres, liteq gives you background jobs, retries, dead-letter queues, and operational controls—without Redis, brokers, or coordination services.

Why liteq

Fewer moving parts
No Redis cluster to maintain, no message broker to monitor. One database, one dependency.

Simple mental model
Create a queue → enqueue tasks → run workers. The entire API fits on one screen.

Production-ready defaults
Exponential backoff retries, dead-letter queues, transactional enqueue, worker timeouts, pause/resume/drain—already built in.

Debuggable when things go wrong
Tasks are rows in Postgres. Query them with SQL. Replay failed jobs from the DLQ. No black boxes.

CLI for operations
Create queues, pause traffic, drain backlogs, inspect history—all from your terminal with lq.

When to use liteq

Good fit:

  • You're already on PostgreSQL and want to avoid adding Redis/RabbitMQ
  • You need reliable background jobs with retries and dead-letter handling
  • You want transactional task enqueue (task commits only if your transaction commits)
  • You value operational simplicity and SQL-based debugging
  • You need cross-language support (any language with a Postgres driver can implement the protocol spec)

Not a fit:

  • You need advanced routing patterns (topic exchanges, fanout, priority queues)
  • You're pushing extreme throughput and latency is critical (sub-millisecond response times)
  • You already have a mature Redis/SQS setup with no operational pain

How it works

Producer → Enqueue(task) → PostgreSQL table → Worker polls → Consume(task)
                                  ↓
                            Retry on error
                                  ↓
                            Max retries → Dead Letter Queue

Tasks are rows in a Postgres table. Workers poll for available tasks, process them with your Consumer, and handle retries/failures automatically. Everything uses transactions—enqueue is atomic with your business logic.

Install

go get github.com/crazzyghost/liteq.go@latest

CLI:

Download pre-built binary from GitHub Releases:

# macOS (Apple Silicon)
curl -sL https://github.com/crazzyghost/liteq.go/releases/download/<version>/lq_darwin_arm64.tar.gz | tar xz
sudo mv lq /usr/local/bin/

# macOS (Intel)
curl -sL https://github.com/crazzyghost/liteq.go/releases/download/<version>/lq_darwin_amd64.tar.gz | tar xz
sudo mv lq /usr/local/bin/

# Linux (x86_64)
curl -sL https://github.com/crazzyghost/liteq.go/releases/download/<version>/lq_linux_amd64.tar.gz | tar xz
sudo mv lq /usr/local/bin/

Or install from source:

go install github.com/crazzyghost/liteq.go/cmd/lq@latest

Requirements: Go 1.25+ • PostgreSQL 12+

Quick start

1. Setup

export LITEQ_DATABASE_URL="postgres://localhost:5432/app?sslmode=disable"
lq queue create email_jobs  # Creates email_jobs + email_jobs_dead_letter

2. Enqueue a task

import lq "github.com/crazzyghost/liteq.go"

pool, _ := pgxpool.New(ctx, os.Getenv("LITEQ_DATABASE_URL"))
jobs, _ := lq.NewPgQueue(ctx, pool, "email_jobs", &lq.RetryPolicy{
    Strategy: lq.StrategyExponential, MaxRetries: 3,
}, lq.WithSchema("liteq"))

tx, txCtx, cancel, _ := jobs.BeginTx(ctx)
defer cancel()

jobs.Enqueue(lq.Task{
    // ID is optional. PostgreSQL generates a UUID when it is omitted.
    Data: map[string]any{"email": "ada@example.com"},
}, tx)

tx.Commit(txCtx)

3. Process tasks

type EmailConsumer struct{}

func (EmailConsumer) Consume(ctx context.Context, task lq.Task) error {
    // Send email using task.Data
    return nil  // or return error to retry
}

worker, _ := lq.NewWorker(ctx, &lq.WorkerConfig{
    TaskQueue:       jobs,
    DeadLetterQueue: dlq,
    GetTaskRetryPolicy: func(t lq.Task) (*lq.RetryPolicy, error) {
        return t.RetryPolicy, nil
    },
})

worker.Run(ctx, func() lq.Consumer { return EmailConsumer{} }, 2*time.Second)

That's it. Worker polls every 2 seconds, processes tasks concurrently, retries on failure, and moves exhausted tasks to the DLQ.

Key features

Transactional enqueue

Tasks commit only if your business transaction commits. No orphaned jobs.

tx, _ := pool.Begin(ctx)
saveUser(tx)  // Your business logic
jobs.Enqueue(task, tx)  // Enqueued atomically
tx.Commit(ctx)
Smart retries

Return an error → task retries with exponential backoff. Return ConsumerError{IsNonTransient: true} → straight to DLQ.

func (c Consumer) Consume(ctx context.Context, task lq.Task) error {
    if missingData(task) {
        return &lq.ConsumerError{IsNonTransient: true}  // Skip retries
    }
    return doWork(task)  // Retries on error
}
Operational controls
lq queue pause email_jobs --reason "maintenance"
lq queue drain email_jobs --reason "clear backlog"
lq queue resume email_jobs
lq queue history email_jobs
Auto-migrate option

Skip the CLI—let the app create its own queues:

jobs, _ := lq.NewPgQueue(ctx, pool, "email_jobs", policy, 
    lq.WithAutoMigrate())
Observability hooks
Hooks: lq.SlogHooks{Logger: slog.Default()}  // Structured logs

Or build custom hooks for metrics/tracing by embedding BaseHooks.

Sane defaults
Setting Default Tweak when
TaskBatchSize 10 You want bigger/smaller poll batches
MaxConcurrency runtime.NumCPU() Downstream has rate limits
TaskTimeout 30s Tasks are faster/slower

CLI reference

Command Purpose
lq queue create <name> Create queue + DLQ
lq queue ls List all queues
lq queue pause <name> --reason <msg> Stop processing
lq queue resume <name> Re-enable queue
lq queue drain <name> --reason <msg> Delete pending tasks
lq queue history <name> Show state changes
lq queue rm <name> --force Delete queue

Flags: --database-url, --schema, --dry-run

Reads from LITEQ_DATABASE_URL and LITEQ_SCHEMA env vars by default.

Core API

Queue setup

lq.NewPgQueue(ctx, pool, queueName, retryPolicy, options...)

Options: WithSchema(name), WithAutoMigrate(), WithDeadLetterQueue(dlqName)

Worker setup

lq.NewWorker(ctx, &WorkerConfig{
    TaskQueue, DeadLetterQueue, GetTaskRetryPolicy, Hooks,
    TaskBatchSize, MaxConcurrency, TaskTimeout,
})
worker.Run(ctx, consumerFactory, pollInterval)

Consumer interface

type Consumer interface {
    Consume(ctx context.Context, task Task) error
}

Task struct

type Task struct {
    ID          string
    Data        map[string]any  // JSON-friendly data
    Status      string
    RetryPolicy *RetryPolicy
    // ... metadata fields
}

Retry policy

type RetryPolicy struct {
    Strategy     RetryStrategy  // StrategyExponential | StrategyLinear
    MaxRetries   int
    RetryDelayMs int64
    MaxDelayMs   int64
}

Queue operations

queue.Enqueue(task, tx)
queue.Dequeue(ctx, limit, lockFor, tx)
queue.Pause(ctx)
queue.Resume(ctx)
queue.Drain(ctx)

Hooks

  • SlogHooks{Logger} - Structured logging via slog
  • BaseHooks{} - Embed and override for custom metrics/tracing

Important notes

  • Task.Data must be JSON-serializable - Use map[string]any or structs, not channels/functions
  • Workers need both queues - Pass the main queue and DLQ to WorkerConfig
  • Names are snake_case - Queue/schema names: letters, numbers, underscores only
  • Transactional safety - Always enqueue within a transaction for atomicity

Comparison to alternatives

liteq Sidekiq/BullMQ AWS SQS
Infrastructure PostgreSQL only Redis required Managed service
Language support Any (protocol spec) Ruby/Node/multi Any (HTTP API)
Transactional enqueue ✅ Yes ❌ No ❌ No
Query tasks with SQL ✅ Yes ❌ No ❌ No
Operational complexity Low (1 DB) Medium (DB + Redis) Low (managed)
Cost model DB only DB + Redis infra Per-request pricing
Best for Transactional jobs High throughput Decoupled services

Choose liteq when you value operational simplicity and transactional guarantees. Implement clients in any language by following the SQL operation contracts.

Documentation

Overview

Package liteq provides a PostgreSQL-backed task queue with retry and dead-letter support.

Index

Constants

View Source 
const (
	StrategyFixed       = "fixed"
	StrategyExponential = "exponential"
	StrategyLinear      = "linear"
)

Retry strategy constants define the back-off algorithms available for task retries.

Variables

View Source 
var ErrQueueDraining = fmt.Errorf("queue is draining")

ErrQueueDraining is returned when an operation is attempted on a draining queue. Use errors.Is(err, ErrQueueDraining) to check for this condition.

View Source 
var ErrQueuePaused = fmt.Errorf("queue is paused")

ErrQueuePaused is returned when an operation is attempted on a paused queue. Use errors.Is(err, ErrQueuePaused) to check for this condition.

Functions

func ParseRetryStrategy 

func ParseRetryStrategy(s string) (string, error)

ParseRetryStrategy validates and normalises a retry strategy string. An empty string is treated as the default exponential strategy.

Types

type BaseHooks 

type BaseHooks struct{}

BaseHooks is a no-op Hooks implementation. Embed it in your own struct to selectively override only the hooks you care about.

func (BaseHooks) OnDLQ 

func (BaseHooks) OnDLQ(context.Context, string, string)

OnDLQ is a no-op implementation.

func (BaseHooks) OnDLQFailed 

func (BaseHooks) OnDLQFailed(context.Context, string, error)

OnDLQFailed is a no-op implementation.

func (BaseHooks) OnDequeue 

func (BaseHooks) OnDequeue(context.Context, string, int)

OnDequeue is a no-op implementation.

func (BaseHooks) OnEnqueue 

func (BaseHooks) OnEnqueue(context.Context, string, string)

OnEnqueue is a no-op implementation.

func (BaseHooks) OnQueueDrained 

func (BaseHooks) OnQueueDrained(context.Context, string)

OnQueueDrained is a no-op implementation.

func (BaseHooks) OnQueuePaused 

func (BaseHooks) OnQueuePaused(context.Context, string)

OnQueuePaused is a no-op implementation.

func (BaseHooks) OnQueueResumed 

func (BaseHooks) OnQueueResumed(context.Context, string)

OnQueueResumed is a no-op implementation.

func (BaseHooks) OnRetry 

func (BaseHooks) OnRetry(context.Context, string, int, time.Time)

OnRetry is a no-op implementation.

func (BaseHooks) OnTaskComplete 

func (BaseHooks) OnTaskComplete(context.Context, string, time.Duration)

OnTaskComplete is a no-op implementation.

func (BaseHooks) OnTaskStart 

func (BaseHooks) OnTaskStart(context.Context, string)

OnTaskStart is a no-op implementation.

type BaseQueue 

type BaseQueue struct {
	Ctx         context.Context
	Schema      string
	QueueName   string
	RetryPolicy *RetryPolicy
	TxTimeout   time.Duration // default 5s; applied to every transaction begin
}

BaseQueue holds the shared configuration for all queue implementations.

func (BaseQueue) QualifiedQueueName 

func (q BaseQueue) QualifiedQueueName() string

QualifiedQueueName returns the schema-qualified queue table name.

type BatchError 

type BatchError struct {
	Total  int          // tasks attempted
	Failed int          // tasks that errored
	Errors []*TaskError // individual failures
}

BatchError aggregates task errors from a single Work() batch.

func (*BatchError) Error 

func (e *BatchError) Error() string

func (*BatchError) Unwrap 

func (e *BatchError) Unwrap() []error

type Condition 

type Condition struct {
	Field    string            // e.g., "data", "id", "status", "deleted_at"
	Keys     []string          // JSON path segments, e.g. ["config", "settings", "theme"]
	JSONText bool              // If true, last key uses ->> (text extraction); if false, uses -> (JSON)
	Operator ConditionOperator // The comparison operator
	Value    any               // nil for IS NULL/IS NOT NULL operators
}

Condition describes a single WHERE-clause filter for queue task queries.

func ColumnEquals 

func ColumnEquals(column string, value any) Condition

ColumnEquals creates a Condition that checks equality on a plain column.

func DataCondition 

func DataCondition(key string, op ConditionOperator, value any) Condition

DataCondition creates a Condition for a JSON text extraction on the data column.

func DataEquals 

func DataEquals(key string, value any) Condition

DataEquals creates a Condition that checks equality on a data JSON key.

func DataIsNotNull 

func DataIsNotNull(key string) Condition

DataIsNotNull creates a Condition that checks a data JSON key is not null.

func DataIsNull 

func DataIsNull(key string) Condition

DataIsNull creates a Condition that checks a data JSON key is null.

func DataJSONCondition 

func DataJSONCondition(keys []string, op ConditionOperator, value any) Condition

DataJSONCondition creates a Condition for a JSON object path on the data column.

func DataNotEquals 

func DataNotEquals(key string, value any) Condition

DataNotEquals creates a Condition that checks inequality on a data JSON key.

func DataPathCondition 

func DataPathCondition(keys []string, op ConditionOperator, value any) Condition

DataPathCondition creates a Condition for a deep JSON text path on the data column.

func DeletedAtIsNull 

func DeletedAtIsNull() Condition

DeletedAtIsNull creates a Condition that filters for non-deleted tasks.

func IDEquals 

func IDEquals(id any) Condition

IDEquals creates a Condition that matches a task by its ID.

func NewCondition 

func NewCondition(field string, op ConditionOperator, value any) Condition

NewCondition creates a Condition for a plain column comparison.

func NewJSONCondition 

func NewJSONCondition(field, key string, op ConditionOperator, value any) Condition

NewJSONCondition creates a Condition that extracts a single JSON key as text.

func NewJSONPathCondition 

func NewJSONPathCondition(field string, keys []string, op ConditionOperator, value any, jsonObj bool) Condition

NewJSONPathCondition creates a Condition that traverses a multi-segment JSON path.

func StatusEquals 

func StatusEquals(status string) Condition

StatusEquals creates a Condition that filters tasks by status.

func (*Condition) ApplyToSelect 

func (c *Condition) ApplyToSelect(query SelectQuery)

ApplyToSelect adds this condition as a WHERE clause to a SELECT query.

func (*Condition) ApplyToUpdate 

func (c *Condition) ApplyToUpdate(query UpdateQuery)

ApplyToUpdate adds this condition as a WHERE clause to an UPDATE query.

type ConditionOperator 

type ConditionOperator string

ConditionOperator represents a SQL comparison or logical operator. 

const (
	OpEqual       ConditionOperator = "="
	OpNotEqual    ConditionOperator = "!="
	OpGreaterThan ConditionOperator = ">"
	OpLessThan    ConditionOperator = "<"
	OpGreaterEq   ConditionOperator = ">="
	OpLessEq      ConditionOperator = "<="
	OpLike        ConditionOperator = "LIKE"
	OpIsNull      ConditionOperator = "IS NULL"
	OpIsNotNull   ConditionOperator = "IS NOT NULL"
	OpIn          ConditionOperator = "IN"
	OpNotIn       ConditionOperator = "NOT IN"
	OpAnd         ConditionOperator = "AND"
	OpOr          ConditionOperator = "OR"
	OpNot         ConditionOperator = "NOT"
	OpExists      ConditionOperator = "EXISTS"
	OpNotExists   ConditionOperator = "NOT EXISTS"
	OpBetween     ConditionOperator = "BETWEEN"
	OpNotBetween  ConditionOperator = "NOT BETWEEN"
	OpILike       ConditionOperator = "ILIKE"
	OpNotILike    ConditionOperator = "NOT ILIKE"
	OpIRegexp     ConditionOperator = "~*"
	OpNotIRegexp  ConditionOperator = "!~*"
	OpRegexp      ConditionOperator = "~"
	OpNotRegexp   ConditionOperator = "!~"
	OpJSONGet     ConditionOperator = "->"
	OpJSONGetText ConditionOperator = "->>"
)

Supported SQL operators for use in Condition filters.

type Consumer 

type Consumer interface {
	Consume(ctx context.Context, task Task) (err error)
}

Consumer processes a single dequeued task.

type ConsumerError 

type ConsumerError struct {
	Source         error
	IsNonTransient bool
}

ConsumerError wraps an error returned by a Consumer with transience metadata.

func (*ConsumerError) Error 

func (e *ConsumerError) Error() string

func (*ConsumerError) Unwrap 

func (e *ConsumerError) Unwrap() error

type ConsumerFactory 

type ConsumerFactory func() Consumer

ConsumerFactory creates a new Consumer instance for each work cycle.

type Hooks 

type Hooks interface {
	OnEnqueue(ctx context.Context, queueName, entryID string)
	OnDequeue(ctx context.Context, queueName string, count int)
	OnTaskStart(ctx context.Context, taskID string)
	OnTaskComplete(ctx context.Context, taskID string, duration time.Duration)
	OnRetry(ctx context.Context, taskID string, attempt int, nextRunAt time.Time)
	OnDLQ(ctx context.Context, taskID, reason string)
	OnDLQFailed(ctx context.Context, taskID string, err error)
	OnQueuePaused(ctx context.Context, queueName string)
	OnQueueResumed(ctx context.Context, queueName string)
	OnQueueDrained(ctx context.Context, queueName string)
}

Hooks defines lifecycle callbacks for observability. Implement this interface to integrate custom logging, metrics, or tracing into the worker. Embed BaseHooks to selectively override only the methods you need.

type MaxRetriesExceededError 

type MaxRetriesExceededError struct {
	Retries    int
	MaxRetries int
}

MaxRetriesExceededError indicates a task has exceeded its maximum retry count.

func (*MaxRetriesExceededError) Error 

func (e *MaxRetriesExceededError) Error() string

type PgQueue 

type PgQueue struct {
	BaseQueue
	Pool *pgxpool.Pool
	// contains filtered or unexported fields
}

PgQueue is a PostgreSQL-backed queue that stores and retrieves entries using pgxpool transactions.

func NewPgQueue 

func NewPgQueue(ctx context.Context, pool *pgxpool.Pool, queueName string, retryPolicy *RetryPolicy, opts ...PgQueueOption) (*PgQueue, error)

NewPgQueue validates its inputs and constructs a PgQueue. A non-nil error is returned when any required argument is missing or the retry policy is invalid.

func (*PgQueue) BeginTx 

func (q *PgQueue) BeginTx(ctx context.Context) (pgx.Tx, context.Context, context.CancelFunc, error)

BeginTx starts a new transaction scoped to the queue's TxTimeout.

func (*PgQueue) CheckCondition 

func (q *PgQueue) CheckCondition(ctx context.Context, tx pgx.Tx, conditions ...Condition) (bool, error)

CheckCondition returns true when at least one row matches all conditions.

func (*PgQueue) Dequeue 

func (q *PgQueue) Dequeue(batchSize int) (tasks []Task, err error)

Dequeue claims up to batchSize pending tasks atomically, setting them to RUNNING.

func (*PgQueue) Drain 

func (q *PgQueue) Drain(ctx context.Context) error

Drain deletes all queue entries and leaves the queue paused.

func (*PgQueue) Enqueue 

func (q *PgQueue) Enqueue(task Task, tx pgx.Tx) error

Enqueue inserts a queue entry into the queue table within the given transaction.

func (*PgQueue) GetRetryPolicy 

func (q *PgQueue) GetRetryPolicy(ctx context.Context) (*RetryPolicy, error)

GetRetryPolicy returns the queue's retry policy, loading it from the database on first access if no static policy was provided at construction time.

func (*PgQueue) IsPaused 

func (q *PgQueue) IsPaused(ctx context.Context) (bool, error)

IsPaused reports whether the queue is currently paused.

func (*PgQueue) Pause 

func (q *PgQueue) Pause(ctx context.Context) error

Pause marks the queue as paused and records the state change.

func (*PgQueue) QueueLabel 

func (q *PgQueue) QueueLabel() string

QueueLabel returns the queue's table name for display/logging purposes.

func (*PgQueue) Resume 

func (q *PgQueue) Resume(ctx context.Context) error

Resume marks the queue as active and records the state change.

func (*PgQueue) Select 

func (q *PgQueue) Select(
	ctx context.Context,
	scan func(pgx.Rows) error,
	mods ...SelectMod,
) error

Select builds and executes a SELECT against the queue table. sm.From is always prepended automatically so callers don't need it. scan is called once per row; returning an error stops iteration immediately. All filtering, ordering, and column selection is expressed as SelectMods (e.g. sm.Where, sm.OrderBy, sm.Limit, sm.Columns).

func (*PgQueue) SelectOne 

func (q *PgQueue) SelectOne(
	ctx context.Context,
	scan func(pgx.Rows) error,
	mods ...SelectMod,
) (bool, error)

SelectOne executes a SELECT and calls scan for the first matching row. Returns (true, nil) when a row was found, (false, nil) when none matched. scan receives a pgx.Rows already positioned on the row — call rows.Scan() directly inside it.

func (*PgQueue) UpdateEntry 

func (q *PgQueue) UpdateEntry(task Task, tx pgx.Tx, conditions ...Condition) error

UpdateEntry persists changes to a queue entry within the given transaction.

func (*PgQueue) UpdateStatus 

func (q *PgQueue) UpdateStatus(ctx context.Context, tx pgx.Tx, status string, conditions ...Condition) error

UpdateStatus sets the status column for rows matching the given conditions.

type PgQueueOption 

type PgQueueOption func(*pgQueueConfig)

PgQueueOption configures a PgQueue during construction.

func WithAutoMigrate 

func WithAutoMigrate() PgQueueOption

WithAutoMigrate enables automatic schema migration on queue creation.

func WithDLQName 

func WithDLQName(name string) PgQueueOption

WithDLQName overrides the DLQ table name used by WithAutoMigrate. When unset, liteq derives "<queue>_dead_letter" for non-DLQ queues.

func WithSchema 

func WithSchema(schema string) PgQueueOption

WithSchema sets the PostgreSQL schema for queue tables.

type Queue 

type Queue interface {
	Enqueue(task Task, tx pgx.Tx) error
	Dequeue(batchSize int) ([]Task, error)
	UpdateEntry(task Task, tx pgx.Tx, conditions ...Condition) error
	CheckCondition(ctx context.Context, tx pgx.Tx, conditions ...Condition) (bool, error)
	Select(ctx context.Context, scan func(pgx.Rows) error, mods ...SelectMod) error
	SelectOne(ctx context.Context, scan func(pgx.Rows) error, mods ...SelectMod) (bool, error)
	UpdateStatus(ctx context.Context, tx pgx.Tx, status string, conditions ...Condition) error
	GetRetryPolicy(ctx context.Context) (*RetryPolicy, error)
	BeginTx(ctx context.Context) (pgx.Tx, context.Context, context.CancelFunc, error)
	QueueLabel() string

	// Queue state management.
	Pause(ctx context.Context) error
	Resume(ctx context.Context) error
	IsPaused(ctx context.Context) (bool, error)
	Drain(ctx context.Context) error
}

Queue defines the stable queue operations supported by liteq backends.

The method set mirrors the current PgQueue API so callers can depend on an interface without forcing a breaking rewrite of the package surface.

type QueueData 

type QueueData interface{}

QueueData is the data payload stored in a queue entry.

type QueueDefinition 

type QueueDefinition struct {
	Name       string
	DLQName    string
	DisableDLQ bool
}

QueueDefinition specifies a queue and its dead-letter queue migration behavior.

type RetryPolicy 

type RetryPolicy struct {
	Strategy     string `json:"strategy"`
	MaxRetries   int    `json:"maxRetries"`
	RetryDelayMs int    `json:"retryDelayMs"`
	MaxDelayMs   int    `json:"maxDelayMs"`
}

RetryPolicy configures the back-off behavior for failed tasks.

type SchemaManager 

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

SchemaManager handles database migrations for liteq queue tables.

func NewSchemaManager 

func NewSchemaManager(pool *pgxpool.Pool, opts ...SchemaManagerOption) *SchemaManager

NewSchemaManager creates a SchemaManager with the given pool and options.

func (*SchemaManager) EnsureQueue 

func (sm *SchemaManager) EnsureQueue(ctx context.Context, name, dlqName string) error

EnsureQueue applies migrations for a single queue and its optional DLQ.

func (*SchemaManager) EnsureSchema 

func (sm *SchemaManager) EnsureSchema(ctx context.Context) error

EnsureSchema applies only the foundation migration (schema + migrations table).

func (*SchemaManager) Migrate 

func (sm *SchemaManager) Migrate(ctx context.Context, queues []QueueDefinition) error

Migrate applies all pending up-migrations for the given queue definitions.

func (*SchemaManager) MigrateDown 

func (sm *SchemaManager) MigrateDown(ctx context.Context, steps int) error

MigrateDown rolls back the given number of migration batches.

func (*SchemaManager) MigrateDownAll 

func (sm *SchemaManager) MigrateDownAll(ctx context.Context) error

MigrateDownAll rolls back all applied migration batches.

func (*SchemaManager) MigrateDownQueue 

func (sm *SchemaManager) MigrateDownQueue(ctx context.Context, queueName string, steps int) error

MigrateDownQueue rolls back the given number of migration batches for a single queue.

func (*SchemaManager) MigrateDownQueueAll 

func (sm *SchemaManager) MigrateDownQueueAll(ctx context.Context, queueName string) error

MigrateDownQueueAll rolls back all applied migration batches for a single queue.

func (*SchemaManager) MigrateWithStatus 

func (sm *SchemaManager) MigrateWithStatus(ctx context.Context, queues []QueueDefinition) (bool, error)

MigrateWithStatus applies all pending up-migrations for the given queue definitions and reports whether any new migration records were applied.

type SchemaManagerOption 

type SchemaManagerOption func(*schemaManagerConfig)

SchemaManagerOption configures a SchemaManager at construction time.

func WithDryRun 

func WithDryRun(w io.Writer) SchemaManagerOption

WithDryRun enables dry-run mode, writing SQL to w instead of executing it.

func WithSchemaManagerSchema 

func WithSchemaManagerSchema(schema string) SchemaManagerOption

WithSchemaManagerSchema sets the target Postgres schema for migrations.

type SelectMod 

type SelectMod = bob.Mod[*dialect.SelectQuery]

SelectMod is a query modifier for PostgreSQL SELECT statements.

type SelectQuery 

type SelectQuery interface {
	Apply(mods ...SelectMod)
}

SelectQuery is a SELECT query builder that accepts SelectMod modifiers.

type SlogHooks 

type SlogHooks struct {
	BaseHooks
	Logger *slog.Logger
}

SlogHooks emits structured log lines via slog for each lifecycle event.

func (SlogHooks) OnDLQ 

func (h SlogHooks) OnDLQ(ctx context.Context, taskID, reason string)

OnDLQ logs a dead-letter queue enqueue event.

func (SlogHooks) OnDLQFailed 

func (h SlogHooks) OnDLQFailed(ctx context.Context, taskID string, err error)

OnDLQFailed logs a dead-letter queue enqueue failure.

func (SlogHooks) OnDequeue 

func (h SlogHooks) OnDequeue(ctx context.Context, queueName string, count int)

OnDequeue logs a batch dequeue event.

func (SlogHooks) OnEnqueue 

func (h SlogHooks) OnEnqueue(ctx context.Context, queueName, entryID string)

OnEnqueue logs a task enqueue event.

func (SlogHooks) OnQueueDrained 

func (h SlogHooks) OnQueueDrained(ctx context.Context, queueName string)

OnQueueDrained logs a queue drain event.

func (SlogHooks) OnQueuePaused 

func (h SlogHooks) OnQueuePaused(ctx context.Context, queueName string)

OnQueuePaused logs a queue pause event.

func (SlogHooks) OnQueueResumed 

func (h SlogHooks) OnQueueResumed(ctx context.Context, queueName string)

OnQueueResumed logs a queue resume event.

func (SlogHooks) OnRetry 

func (h SlogHooks) OnRetry(ctx context.Context, taskID string, attempt int, nextRunAt time.Time)

OnRetry logs a task retry scheduling event.

func (SlogHooks) OnTaskComplete 

func (h SlogHooks) OnTaskComplete(ctx context.Context, taskID string, duration time.Duration)

OnTaskComplete logs a task completion event with duration.

func (SlogHooks) OnTaskStart 

func (h SlogHooks) OnTaskStart(ctx context.Context, taskID string)

OnTaskStart logs a task start event.

type Task 

type Task struct {
	ID          string       `json:"id" db:"id"`
	Data        QueueData    `json:"data" db:"data"`
	Status      string       `json:"status" db:"status"`
	IsRetry     bool         `json:"isRetry" db:"is_retry"`
	Retries     int          `json:"retries" db:"retries"`
	RetryPolicy *RetryPolicy `json:"retryPolicy" db:"retry_policy"`
	NextRunAt   *time.Time   `json:"nextRunAt" db:"next_run_at"`
	LastRunAt   *time.Time   `json:"lastRunAt" db:"last_run_at"`
	ProcessedAt *time.Time   `json:"processedAt" db:"processed_at"`
	EnqueuedAt  *time.Time   `json:"enqueued_at" db:"enqueued_at"`
	DequeuedAt  *time.Time   `json:"dequeued_at" db:"dequeued_at"`
	CreatedAt   *time.Time   `json:"created_at" db:"created_at"`
	UpdatedAt   *time.Time   `json:"updated_at" db:"updated_at"`
	DeletedAt   *time.Time   `json:"deleted_at" db:"deleted_at"`
}

Task is a queue entry representing a unit of work to be processed. ID stores the canonical UUID string for the queue row. Leave it empty on enqueue to let PostgreSQL generate a new UUID.

func (*Task) WillExceedMaxRetries 

func (t *Task) WillExceedMaxRetries(maxRetries int) bool

WillExceedMaxRetries reports whether the next retry would exceed maxRetries.

type TaskError 

type TaskError struct {
	TaskID string
	Err    error
}

TaskError wraps a single task processing failure.

func (*TaskError) Error 

func (e *TaskError) Error() string

func (*TaskError) Unwrap 

func (e *TaskError) Unwrap() error

type TaskRetryPolicySelector 

type TaskRetryPolicySelector func(task Task) (*RetryPolicy, error)

TaskRetryPolicySelector resolves the retry policy for a given task. Returning (nil, nil) indicates that no retries are allowed.

type TaskStatus 

type TaskStatus string

TaskStatus represents the lifecycle state of a task in the queue. 

const (
	PENDING   TaskStatus = "PENDING"
	RUNNING   TaskStatus = "RUNNING"
	FAILED    TaskStatus = "FAILED"
	COMPLETED TaskStatus = "COMPLETED"
	CANCELLED TaskStatus = "CANCELLED" //nolint:misspell // CANCELLED is the stored DB value
	DLQFailed TaskStatus = "DLQ_FAILED"
)

Task status constants represent all possible states in a task's lifecycle.

type UpdateMod 

type UpdateMod = bob.Mod[*dialect.UpdateQuery]

UpdateMod is a query modifier for PostgreSQL UPDATE statements.

type UpdateQuery 

type UpdateQuery interface {
	Apply(mods ...UpdateMod)
}

UpdateQuery is an UPDATE query builder that accepts UpdateMod modifiers.

type Worker 

type Worker struct {
	Ctx                context.Context
	TaskBatchSize      int
	TaskQueue          Queue
	DeadLetterQueue    Queue
	GetTaskRetryPolicy TaskRetryPolicySelector
	MaxConcurrency     int
	TaskTimeout        time.Duration
	// contains filtered or unexported fields
}

Worker dequeues tasks from a queue, processes them via a Consumer, and handles retries and dead-letter enqueueing on failure.

func NewWorker 

func NewWorker(ctx context.Context, config *WorkerConfig) (*Worker, error)

NewWorker validates config and constructs a Worker. A non-nil error is returned when any required field is missing or has an invalid value.

func (*Worker) DlqEnqueue 

func (w *Worker) DlqEnqueue(ctx context.Context, task *Task, tx pgx.Tx) (err error)

DlqEnqueue sends a failed task to the dead-letter queue within the given transaction.

func (*Worker) GetRetrySchedule 

func (w *Worker) GetRetrySchedule(task *Task, retryPolicy *RetryPolicy) (time.Time, error)

GetRetrySchedule calculates the next run time for a task based on the retry policy strategy. Supported strategies are fixed, linear, and exponential (default). A jitter in the range [0, delayMs] is applied to spread load.

func (*Worker) HandleFailure 

func (w *Worker) HandleFailure(ctx context.Context, task *Task, failure error, tx pgx.Tx, start time.Time) error

HandleFailure marks a task as failed and either retries or dead-letters it.

func (*Worker) HandleProcessed 

func (w *Worker) HandleProcessed(ctx context.Context, task *Task, status TaskStatus, tx pgx.Tx, start time.Time) (err error)

HandleProcessed persists the task's new status and fires completion hooks.

func (*Worker) HandleUnit 

func (w *Worker) HandleUnit(ctx context.Context, task *Task, consumer Consumer) error

HandleUnit runs the consumer for one task with a deadline, then commits the result (completed, retry, or DLQ) inside a transaction. A timeout is treated as a transient failure and follows the normal retry path.

func (*Worker) Poll 

func (w *Worker) Poll(ctx context.Context) (tasks []Task, err error)

Poll dequeues one batch of tasks from the task queue.

func (*Worker) Retry 

func (w *Worker) Retry(ctx context.Context, task *Task, tx pgx.Tx) (err error)

Retry enqueues the task for another attempt according to its retry policy.

func (*Worker) Run 

func (w *Worker) Run(ctx context.Context, factory ConsumerFactory, interval time.Duration) error

Run polls in a loop until Stop() is called or ctx is canceled. Errors from individual Work() batches are logged but do not abort the loop.

func (*Worker) Stop 

func (w *Worker) Stop()

Stop signals the worker to drain and exit. Safe to call multiple times.

func (*Worker) Work 

func (w *Worker) Work(ctx context.Context, factory ConsumerFactory) error

Work dequeues one batch and processes all tasks using a bounded worker pool. All tasks complete regardless of individual failures; errors are aggregated into a BatchError.

type WorkerConfig 

type WorkerConfig struct {
	TaskBatchSize      int
	TaskQueue          Queue
	DeadLetterQueue    Queue
	GetTaskRetryPolicy TaskRetryPolicySelector
	MaxConcurrency     int           // default: runtime.NumCPU()
	TaskTimeout        time.Duration // default: 30s
	Hooks              Hooks         // default: BaseHooks{} (silent)
}

WorkerConfig holds the configuration used to construct a Worker.

Source Files

View all Source files

Directories

Path Synopsis
cmd
lq command
Package main implements the lq CLI for queue administration.
 Package main implements the lq CLI for queue administration.

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.