goqueue

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Jan 19, 2026 License: GPL-3.0 Imports: 17 Imported by: 0

README

goqueue

License Go Report Card

A simple, efficient, and reliable queue implementation for Go with PostgreSQL backend support.

Table of Contents

Features

  • Concurrent Processing: Efficient worker-based job processing
  • FIFO Queues: First-In-First-Out job queues for strict ordering requirements
  • Retry Mechanism: Customizable retry logic for failed jobs
  • Dead Letter Queue: Automatic handling of permanently failed jobs for later analysis
  • PostgreSQL Backend: Reliable, persistent job storage
  • Structured Logging: Built-in support for Go's slog package

Getting Started

Prerequisites
  • Go 1.21 or higher
  • PostgreSQL 13 or higher
  • sqlc for generating type-safe SQL bindings
  • Make for build automation
Installation
git clone https://github.com/flohansen/goqueue
cd goqueue

Development

Available Commands
Command Description
make generate Generate Go code from SQL files (required after any .sql file changes)
make test Run all unit tests
make test-it Run all integration tests
Testing
Unit Tests

Unit tests follow Go conventions and are located alongside implementation files:

.
├── queue.go
├── queue_test.go
└── ...

Run unit tests:

make test

Or manually:

go test ./... -cover -v
Integration Tests

Integration tests are located in tests/integration/ and require the integration build tag:

//go:build integration

Run integration tests:

make test-it

Or manually:

go test -tags=integration ./test/integration/... -v
Working with SQL
Migrations

SQL migration files are stored in sql/migrations/ and follow this naming convention:

<version>_<name>.[up|down].sql

Example: 000001_create_jobs_table.up.sql

Queries

SQL query files are stored in sql/queries/. Go bindings are automatically generated from these files using sqlc.

Important: After modifying any .sql file, regenerate the Go bindings:

make generate

Dead Letter Queue

The dead letter queue (DLQ) feature provides a robust way to handle jobs that fail after exhausting all retry attempts. Jobs that cannot be successfully processed are automatically moved to a dead letter queue for later analysis and debugging.

How It Works
  1. When a job fails, it enters the retry mechanism based on your configured retry policy
  2. After the maximum number of retries is reached, the job is moved to the dead letter queue
  3. Jobs in the DLQ can be queried, analyzed, and reprocessed if needed
  4. This prevents failed jobs from being lost and allows for post-mortem analysis
Benefits
  • Observability: Track jobs that fail unexpectedly
  • Debugging: Analyze failed jobs to identify issues in your processing logic
  • Recovery: Reprocess failed jobs once underlying issues are resolved
  • Data Integrity: Ensure no job is silently dropped due to failures

Using Nix (Optional)

This repository includes a flake.nix for simplified dependency management. With Nix installed, you can:

Spawn a development shell with all dependencies:

nix develop

Or run commands directly:

nix develop --command make generate

Contributing

Contributions are welcome! Here's how you can help:

  1. Fork the repository
  2. Create a feature branch (git checkout -b feature/amazing-feature)
  3. Make your changes and add tests
  4. Run tests to ensure everything works (make test && make test-it)
  5. Commit your changes (git commit -m 'feat: add amazing feature')
  6. Push to your branch (git push origin feature/amazing-feature)
  7. Open a Pull Request

Please ensure your code follows Go best practices and includes appropriate tests.

License

This project is licensed under the GPL-3.0 License. See the LICENSE file for details.

Questions or Issues? Feel free to open an issue on GitHub.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func MigrateUp 

func MigrateUp(pool *pgxpool.Pool) error

MigrateUp applies all up migrations to the database connected via the given pgxpool.Pool.

Types

type DB 

type DB interface {
	database.DBTX

	// BeginTx creates a new (pgx) transaction
	BeginTx(ctx context.Context, options pgx.TxOptions) (pgx.Tx, error)
}

DB is the abstraction for any (pgx) database connection with allowing to create transactions.

type EnqueueHandler 

type EnqueueHandler func(ctx JobEnqueueContext) error

EnqueueHandler defines the function signature for handling job enqueuing.

type EnqueueOption 

type EnqueueOption func(*enqueueConfig)

EnqueueOption defines a functional option for configuring job enqueuing.

func WithMaxRetries 

func WithMaxRetries(n int32) EnqueueOption

WithMaxRetries sets the maximum number of retries for the enqueued job.

func WithRetryPolicy 

func WithRetryPolicy(policy RetryPolicy) EnqueueOption

WithRetryPolicy sets the retry policy for the enqueued job. Possible policies are RetryPolicyConstant, RetryPolicyLinear, and RetryPolicyExponential.

type Job 

type Job[T any] struct {
	// ID is the database identifier for the job.
	ID int32

	// Metadata contains meta information of a job stored as key-value pairs.
	Metadata map[string]any

	// Args contains the job payload of type T.
	Args T
}

Job represents a queued job with its arguments and database ID.

type JobEnqueueContext 

type JobEnqueueContext struct {
	context.Context
	Metadata map[string]any
	Args     any
}

JobEnqueueContext provides context for enqueuing a job, including its arguments.

func (JobEnqueueContext) WithContext 

func (ectx JobEnqueueContext) WithContext(ctx context.Context) JobEnqueueContext

WithContext returns a copy of the JobEnqueueContext with the provided context.

type JobProcessContext 

type JobProcessContext struct {
	context.Context
	JobID    int32
	Metadata map[string]any
	Args     any
}

JobProcessContext provides context for processing a job, including its ID and arguments.

func (JobProcessContext) WithContext 

func (pctx JobProcessContext) WithContext(ctx context.Context) JobProcessContext

WithContext returns a copy of the JobProcessContext with the provided context.

type JobQueue 

type JobQueue[T any] struct {
	// contains filtered or unexported fields
}

JobQueue manages the lifecycle of jobs in a named queue. It polls the database for scheduled jobs, executes them via a Worker, and handles retries with exponential backoff or other configured policies. Jobs are processed concurrently.

T is the type of the job arguments payload and must be JSON serializable.

JobQueues should be created using the New function. Example: 

queue := goqueue.New(db, &MyWorker{},
    goqueue.WithQueueName("my-queue"),
    goqueue.WithPollInterval(500*time.Millisecond),
)

func New 

func New[T any](db DB, worker Worker[T], opts ...JobQueueOption) *JobQueue[T]

New creates a new JobQueue with the given database connection and worker. Configure behavior using functional options like WithPollInterval, WithQueueName, etc.

func (*JobQueue[T]) Enqueue 

func (jq *JobQueue[T]) Enqueue(ctx context.Context, args T, opts ...EnqueueOption) (*Job[T], error)

Enqueue adds a new job with the given arguments to the queue. Optional parameters can be set using EnqueueOption functions like WithMaxRetries and WithRetryPolicy. It returns the created Job or an error. Example: 

job, err := queue.Enqueue(ctx, MyJobArgs{...},
    jobqueue.WithMaxRetries(5),
    jobqueue.WithRetryPolicy(jobqueue.RetryPolicyExponential),
)

func (*JobQueue[T]) Receive 

func (jq *JobQueue[T]) Receive(ctx context.Context)

Receive starts polling the queue for scheduled jobs and processes them using the configured Worker. It runs until the provided context is cancelled. Errors during job processing are logged but do not stop the polling loop. Failed jobs are retried or marked as failed after the retry policy has been exhausted.

Example: 

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

go queue.Receive(ctx)

// Run for 10 minutes
time.Sleep(10 * time.Minute)
cancel()

type JobQueueOption 

type JobQueueOption func(*jobQueueConfig)

JobQueueOption defines a functional option for configuring a JobQueue.

func WithBaseRetryDelay 

func WithBaseRetryDelay(d time.Duration) JobQueueOption

WithBaseRetryDelay sets the base delay used for calculating retry backoff.

func WithDLQ 

func WithDLQ(queueName string) JobQueueOption

WithDLQ configures a dead-letter queue (DLQ) for handling failed jobs. If queueName is an empty string, no DLQ is used.

func WithFIFO 

func WithFIFO(isFIFO bool) JobQueueOption

WithFIFO configures the queue to process jobs in a first-in-first-out manner.

func WithLogger 

func WithLogger(logger *slog.Logger) JobQueueOption

WithLogger sets a custom logger for the JobQueue.

func WithMaxRetryDelay 

func WithMaxRetryDelay(d time.Duration) JobQueueOption

WithMaxRetryDelay sets the maximum delay allowed between retry attempts.

func WithMiddlewares 

func WithMiddlewares(mws ...Middleware) JobQueueOption

WithMiddlewares adds the given middlewares to the JobQueue for both enqueuing and processing jobs. They are applied in the order provided.

func WithPollInterval 

func WithPollInterval(interval time.Duration) JobQueueOption

WithPollInterval sets the polling interval for checking the queue for new jobs.

func WithQueueName 

func WithQueueName(name string) JobQueueOption

WithQueueName sets the name of the job queue.

type Middleware 

type Middleware interface {
	// Enqueue is called when a job is being enqueued.
	Enqueue(next EnqueueHandler) EnqueueHandler

	// Process is called when a job is being processed.
	Process(next ProcessHandler) ProcessHandler
}

Middleware allows wrapping EnqueueHandler and ProcessHandler with custom logic.

type ProcessHandler 

type ProcessHandler func(ctx JobProcessContext) error

ProcessHandler defines the function signature for handling job processing.

type Querier 

type Querier interface {
	database.Querier
	WithTx(tx pgx.Tx) *database.Queries
}

Querier combines the generated database queries with the ability to create a new Querier bound to a transaction.

type RetryPolicy 

type RetryPolicy int

RetryPolicy defines the strategy for retrying failed jobs. 

const (
	// RetryPolicyConstant retries jobs with a constant delay. Formlar:
	//
	//	next_delay = base_delay
	//
	RetryPolicyConstant RetryPolicy = iota

	// RetryPolicyLinear retries jobs with a linearly increasing delay. Formula:
	//
	//	next_delay = base_delay * (attempt_number + 1)
	//
	RetryPolicyLinear

	// RetryPolicyExponential retries jobs with an exponentially increasing delay. Formula:
	//
	//	next_delay = base_delay ^ (attempt_number + 1)
	//
	RetryPolicyExponential
)

type UnimplementedMiddleware 

type UnimplementedMiddleware struct {
	Middleware
}

UnimplementedMiddleware can be embedded to have a Middleware implementation with no methods.

func (*UnimplementedMiddleware) Enqueue 

func (m *UnimplementedMiddleware) Enqueue(next EnqueueHandler) EnqueueHandler

Enqueue returns the next EnqueueHandler unchanged.

func (*UnimplementedMiddleware) Process 

func (m *UnimplementedMiddleware) Process(next ProcessHandler) ProcessHandler

Process returns the next ProcessHandler unchanged.

type Worker 

type Worker[T any] interface {
	// Work processes a single job. Returning an error triggers retry logic
	// based on the job's retry policy and max retry count. If max retries is
	// exceeded, the job is marked as failed.
	Work(ctx context.Context, job *Job[T]) error
}

Worker processes jobs ob type T. Implementations should be idempotent as jobs may be retried multiple times on failure.

Source Files

View all Source files

Directories

Path Synopsis
internal
database
errors

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.