Merge pull request #5 from acaloiaro/add-in-memory-backend

Add in-memory backend

README.md

@@ -1,6 +1,6 @@
 # Neoq
 
-Background job processing built on Postgres for Go
+Background job processing for Go
 
 [![Go Reference](https://pkg.go.dev/badge/github.com/acaloiaro/neoq.svg)](https://pkg.go.dev/github.com/acaloiaro/neoq) [![Gitter chat](https://badges.gitter.im/gitterHQ/gitter.png)](https://app.gitter.im/#/room/#neoq:gitter.im)
 
@@ -10,15 +10,15 @@ Background job processing built on Postgres for Go
 
 # About
 
-Neoq is a background job framework for Go applications. Its purpose is to minimize the infrastructure necessary to run production applications. It does so by implementing queue durability with modular backends, rather than introducing a strict dependency on a particular backend such as Redis.
+Neoq is a background job framework for Go applications. Its purpose is to minimize the infrastructure necessary to run production applications. It does so by implementing queue durability with modular backends.
 
-The initial backend is based on Postgres since many applications already require a ralational database, and Postgres is an excellent one.
+This allows application to use the same type of data store for both application data and backround job processing. At the moment an in-memory and Postgres backends are provided. However, the goal is to have backends for every major datastore: Postgres, Redis, MySQL, etc.
 
 Neoq does not aim to be the _fastest_ background job processor. It aims to be _fast_, _reliable_, and demand a minimal infrastructure footprint.
 
-# Features
+# What it does
 
-- **Postgres-backed** job processing
+- **Background job Processing**: Neoq has an in-memory and Postgres backend out of the box. Users may supply their own without changing neoq directly.
 - **Retries**: Jobs may be retried a configurable number of times with exponential backoff and jitter to prevent thundering herds
 - **Job uniqueness**: jobs are fingerprinted based on their payload and status to prevent job duplication (multiple unprocessed jobs with the same payload cannot be queued)
 - **Deadlines**: Queue handlers can be configured with per-job time deadlines with millisecond accuracy
@@ -37,15 +37,15 @@ Error handling in this section is excluded for simplicity.
 
 ## Add queue handlers
 
-Queue handlers listen for Jobs on queues. Jobs may consist of any payload that is JSON-serializable. Payloads are stored in Postgres in a `jsonb` field.
+Queue handlers listen for Jobs on queues. Jobs may consist of any payload that is JSON-serializable.
 
 Queue Handlers are simple Go functions that accept a `Context` parameter.
 
 **Example**: Add a listener on the `hello_world` queue
 
 ```go
-nq, err := neoq.New(ctx, neoq.ConnectionString("postgres://postgres:postgres@localhost:5432/neoq"))
-handleErr(err)
+ctx := context.Background()
+nq, _ := neoq.New(ctx)
 nq.Listen(ctx, "hello_world", neoq.NewHandler(func(ctx context.Context) (err error) {
   j, err := neoq.JobFromContext(ctx)
   log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
@@ -58,16 +58,14 @@ nq.Listen(ctx, "hello_world", neoq.NewHandler(func(ctx context.Context) (err err
 **Example**: Add a "Hello World" job to the `hello_world` queue
 
 ```go
-nq, err := neoq.New(ctx, neoq.ConnectionString("postgres://postgres:postgres@localhost:5432/neoq"))
-handleErr(err)
-
-jid, err := nq.Enqueue(ctx, neoq.Job{
+ctx := context.Background()
+nq, _ := neoq.New(ctx)
+jid, _ := nq.Enqueue(ctx, neoq.Job{
   Queue: "hello_world",
   Payload: map[string]interface{}{
     "message": "hello world",
   },
 })
-handleErr(err)
 ```
 
 # Example Code

examples/add_future_postgres_job/main.go

@@ -11,9 +11,13 @@ import (
 func main() {
 	const queue = "foobar"
 	ctx := context.Background()
-	nq, err := neoq.New(ctx, neoq.PgTransactionTimeout(1000))
+	nq, err := neoq.New(ctx,
+		neoq.PgTransactionTimeout(1000),
+		neoq.BackendName("postgres"),
+		neoq.ConnectionString("postgres://postgres:postgres@127.0.0.1:5432/neoq"))
+
 	if err != nil {
-		log.Fatalf("error initializing neoq: %v", err)
+		log.Fatalf("error initializing neoq: %w", err)
 	}
 
 	// Add a job that will execute 1 hour from now

examples/add_job_with_custom_concurrency/main.go

@@ -11,8 +11,7 @@ func main() {
 	var err error
 	const queue = "foobar"
 	ctx := context.Background()
-	//
-	nq, err := neoq.New(ctx, neoq.ConnectionString("postgres://postgres:postgres@127.0.0.1:5432/neoq?sslmode=disable"))
+	nq, err := neoq.New(ctx)
 	if err != nil {
 		log.Fatalf("error initializing neoq: %v", err)
 	}
@@ -29,10 +28,10 @@ func main() {
 		log.Println("got job id:", j.ID, "messsage:", j.Payload["message"])
 		done <- true
 		return
-	}, neoq.HandlerConcurreny(8))
+	}, neoq.HandlerConcurrency(8))
 
 	// Option 2: Set options after the handler is created
-	handler = handler.WithOption(neoq.HandlerConcurreny(8))
+	handler = handler.WithOption(neoq.HandlerConcurrency(8))
 
 	err = nq.Listen(ctx, queue, handler)
 	if err != nil {

examples/add_periodic_jobs/main.go

@@ -10,10 +10,7 @@ import (
 
 func main() {
 	ctx := context.Background()
-	// by default neoq connects to a local postgres server using: [neoq.DefaultPgConnectionString]
-	// connection strings can be set explicitly as follows:
-	// neoq.New(neoq.ConnectionString("postgres://username:passsword@hostname/database"))
-	nq, err := neoq.New(ctx, neoq.PgTransactionTimeout(1000))
+	nq, err := neoq.New(ctx)
 	if err != nil {
 		log.Fatalf("error initializing neoq: %v", err)
 	}
@@ -25,7 +22,7 @@ func main() {
 	})
 	handler = handler.
 		WithOption(neoq.HandlerDeadline(500 * time.Millisecond)).
-		WithOption(neoq.HandlerConcurreny(1))
+		WithOption(neoq.HandlerConcurrency(1))
 
 	nq.ListenCron(ctx, "* * * * * *", handler)
 

examples/listen_for_jobs/main.go

@@ -11,8 +11,9 @@ func main() {
 	var err error
 	const queue = "foobar"
 	ctx := context.Background()
-	//
-	nq, err := neoq.New(ctx, neoq.ConnectionString("postgres://postgres:postgres@127.0.0.1:5432/neoq"))
+	nq, err := neoq.New(ctx,
+		neoq.BackendName("postgres"),
+		neoq.ConnectionString("postgres://postgres:postgres@127.0.0.1:5432/neoq"))
 	if err != nil {
 		log.Fatalf("error initializing neoq: %v", err)
 	}

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/iancoleman/strcase v0.2.0
 	github.com/jackc/pgx/v5 v5.3.0
 	github.com/jsuar/go-cron-descriptor v0.1.0
-	github.com/pkg/errors v0.9.1
+	github.com/pkg/errors v0.8.1
 	github.com/robfig/cron v1.2.0
 	golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb
 )

go.sum

@@ -23,9 +23,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
-github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=

memory_backend.go

@@ -0,0 +1,282 @@
+package neoq
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/guregu/null"
+	"github.com/iancoleman/strcase"
+	"github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
+	"github.com/pkg/errors"
+	"github.com/robfig/cron"
+	"golang.org/x/exp/slog"
+)
+
+const (
+	DefaultQueueCapacity = 10000 // the default capacity of individual queues
+	EmptyCapacity        = 0
+)
+
+// MemBackend is a memory-backed neoq backend
+type MemBackend struct {
+	cron         *cron.Cron
+	logger       Logger
+	mu           *sync.Mutex // mutext to protect mutating state on a pgWorker
+	jobCount     int64       // number of jobs that have been queued since start
+	handlers     sync.Map    // map queue names [string] to queue handlers [Handler]
+	fingerprints sync.Map    // map fingerprints [string] to their jobs [Job]
+	futureJobs   sync.Map    // map jobIDs [int64] to [Jobs]
+	queues       sync.Map    // map queue names [string] to queue handler channels [chan Job]
+}
+
+func NewMemBackend(opts ...ConfigOption) (n Neoq, err error) {
+	mb := &MemBackend{
+		cron:         cron.New(),
+		mu:           &sync.Mutex{},
+		queues:       sync.Map{},
+		handlers:     sync.Map{},
+		futureJobs:   sync.Map{},
+		fingerprints: sync.Map{},
+		logger:       slog.New(slog.NewTextHandler(os.Stdout)),
+		jobCount:     0,
+	}
+	mb.cron.Start()
+
+	for _, opt := range opts {
+		opt(mb)
+	}
+
+	n = mb
+
+	return
+}
+
+// Enqueue queues jobs to be executed asynchronously
+func (m *MemBackend) Enqueue(ctx context.Context, job Job) (jobID int64, err error) {
+	var queueChan chan Job
+	var qc any
+	var ok bool
+
+	if qc, ok = m.queues.Load(job.Queue); !ok {
+		return UnqueuedJobID, fmt.Errorf("queue has no listeners: %w", job.Queue)
+	}
+
+	queueChan = qc.(chan Job)
+
+	// Make sure RunAfter is set to a non-zero value if not provided by the caller
+	// if already set, schedule the future job
+	now := time.Now()
+	if job.RunAfter.IsZero() {
+		job.RunAfter = now
+	}
+
+	if job.Queue == "" {
+		err = errors.New("this job does not specify a Queue. Please specify a queue")
+
+		return
+	}
+
+	err = fingerprintJob(&job)
+	if err != nil {
+		return
+	}
+
+	// if the job fingerprint is already known, don't queue the job
+	if _, found := m.fingerprints.Load(job.Fingerprint); found {
+		return DuplicateJobID, nil
+	}
+
+	m.fingerprints.Store(job.Fingerprint, job)
+	m.mu.Lock()
+	m.jobCount++
+	m.mu.Unlock()
+
+	job.ID = m.jobCount
+	jobID = m.jobCount
+
+	// notify listeners that a new job has arrived if it's not a future job
+	if job.RunAfter == now {
+		queueChan <- job
+	} else {
+		m.queueFutureJob(job)
+	}
+
+	return
+}
+
+// Listen listens for jobs on a queue and processes them with the given handler
+func (m *MemBackend) Listen(ctx context.Context, queue string, h Handler) (err error) {
+	var queueCapacity = h.queueCapacity
+	if queueCapacity == EmptyCapacity {
+		queueCapacity = DefaultQueueCapacity
+	}
+
+	m.handlers.Store(queue, h)
+	m.queues.Store(queue, make(chan Job, queueCapacity))
+
+	err = m.start(ctx, queue)
+	if err != nil {
+		return
+	}
+
+	return
+}
+
+// ListenCron listens for jobs on a cron schedule and handles them with the provided handler
+//
+// See: https://pkg.go.dev/github.com/robfig/cron?#hdr-CRON_Expression_Format for details on the cron spec format
+func (m *MemBackend) ListenCron(ctx context.Context, cronSpec string, h Handler) (err error) {
+	cd, err := crondescriptor.NewCronDescriptor(cronSpec)
+	if err != nil {
+		return err
+	}
+
+	cdStr, err := cd.GetDescription(crondescriptor.Full)
+	if err != nil {
+		return err
+	}
+
+	queue := stripNonAlphanum(strcase.ToSnake(*cdStr))
+
+	err = m.Listen(ctx, queue, h)
+	if err != nil {
+		return
+	}
+
+	m.cron.AddFunc(cronSpec, func() {
+		m.Enqueue(ctx, Job{Queue: queue})
+	})
+
+	return
+}
+
+// Shutdown halts the worker
+func (m *MemBackend) Shutdown(ctx context.Context) (err error) {
+	return
+}
+
+// WithConfig configures neoq with with optional configuration
+func (m *MemBackend) WithConfig(opt ConfigOption) (n Neoq) {
+	return
+}
+
+// start starts a queue listener, processes pending job, and fires up goroutines to process future jobs
+func (m *MemBackend) start(ctx context.Context, queue string) (err error) {
+	var queueChan chan Job
+	var qc any
+	var h any
+	var handler Handler
+	var ok bool
+	if h, ok = m.handlers.Load(queue); !ok {
+		return fmt.Errorf("no handler for queue: %s", queue)
+	}
+
+	if qc, ok = m.queues.Load(queue); !ok {
+		return fmt.Errorf("no listener configured for qeuue: %s", queue)
+	}
+
+	go func() { m.scheduleFutureJobs(ctx, queue) }()
+
+	handler = h.(Handler)
+	queueChan = qc.(chan Job)
+
+	for i := 0; i < handler.concurrency; i++ {
+		go func() {
+			var err error
+			var job Job
+
+			for {
+				select {
+				case job = <-queueChan:
+					err = m.handleJob(ctx, job, handler)
+				case <-ctx.Done():
+					return
+				}
+
+				if err != nil {
+					m.logger.Error("error handling job", err, "job_id", job.ID)
+					runAfter := calculateBackoff(job.Retries)
+					job.RunAfter = runAfter
+					m.queueFutureJob(job)
+				}
+
+				m.fingerprints.Delete(job.Fingerprint)
+			}
+		}()
+	}
+	return
+}
+
+func (m *MemBackend) scheduleFutureJobs(ctx context.Context, queue string) {
+	// check for new future jobs on an interval
+	// TODO make this time configurable
+	ticker := time.NewTicker(5 * time.Second)
+
+	for {
+		// loop over list of future jobs, scheduling goroutines to wait for jobs that are due within the next 30 seconds
+		// TODO: Make 30 seconds configurable
+		m.futureJobs.Range(func(k, v any) bool {
+			job := v.(Job)
+			var queueChan chan Job
+
+			at := time.Until(job.RunAfter)
+			if at <= time.Duration(30*time.Second) {
+				m.removeFutureJob(job.ID)
+				go func(j Job) {
+					scheduleCh := time.After(at)
+					<-scheduleCh
+					if qc, ok := m.queues.Load(queue); ok {
+						queueChan = qc.(chan Job)
+						queueChan <- j
+					} else {
+						m.logger.Error(fmt.Sprintf("no listen channel configured for queue: %s", queue), errors.New("no listener configured"))
+					}
+				}(job)
+			}
+
+			return true
+		})
+		select {
+		case <-ticker.C:
+			continue
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (m *MemBackend) handleJob(ctx context.Context, job Job, handler Handler) (err error) {
+	ctxv := handlerCtxVars{job: &job}
+	hctx := withHandlerContext(ctx, ctxv)
+
+	// check if the job is being retried and increment retry count accordingly
+	if job.Status != JobStatusNew {
+		job.Retries = job.Retries + 1
+	}
+
+	// execute the queue handler of this job
+	handlerErr := execHandler(hctx, handler)
+	if handlerErr != nil {
+		job.Error = null.StringFrom(handlerErr.Error())
+	}
+
+	return
+}
+
+// queueFutureJob queues a future job for eventual execution
+func (m *MemBackend) queueFutureJob(job Job) {
+	m.fingerprints.Store(job.Fingerprint, job)
+	m.futureJobs.Store(job.ID, job)
+}
+
+// removeFutureJob removes a future job from the in-memory list of jobs that will execute in the future
+func (m *MemBackend) removeFutureJob(jobID int64) {
+	if j, ok := m.futureJobs.Load(jobID); ok {
+		job := j.(Job)
+		m.fingerprints.Delete(job.Fingerprint)
+		m.futureJobs.Delete(job.ID)
+	}
+}

memory_backend_test.go

@@ -0,0 +1,191 @@
+package neoq
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/pkg/errors"
+	"golang.org/x/exp/slog"
+)
+
+// TestMemeoryBackendBasicJobProcessing tests that the memory backend is able to process the most basic jobs with the
+// most basic configuration.
+func TestMemeoryBackendBasicJobProcessing(t *testing.T) {
+	queue := "testing"
+	numJobs := 1000
+	doneCnt := 0
+	done := make(chan bool)
+	var timeoutTimer = time.After(5 * time.Second)
+
+	ctx := context.TODO()
+	backend, err := NewMemBackend()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	nq, err := New(ctx, Backend(backend))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	handler := NewHandler(func(ctx context.Context) (err error) {
+		done <- true
+		return
+	})
+
+	nq.Listen(ctx, queue, handler)
+
+	go func() {
+		for i := 0; i < numJobs; i++ {
+			jid, err := nq.Enqueue(ctx, Job{
+				Queue: queue,
+				Payload: map[string]interface{}{
+					"message": fmt.Sprintf("hello world: %d", i),
+				},
+			})
+			if err != nil || jid == DuplicateJobID {
+				slog.Error("job was not enqueued. either it was duplicate or this error caused it:", err)
+			}
+		}
+	}()
+
+	for {
+		select {
+		case <-timeoutTimer:
+			err = errors.New("timed out waiting for job(s)")
+		case <-done:
+			doneCnt++
+		}
+
+		if doneCnt >= numJobs {
+			break
+		}
+
+		if err != nil {
+			break
+		}
+	}
+
+	if err != nil {
+		t.Error(err)
+	}
+
+	nq.Shutdown(ctx)
+}
+
+// TestMemoryBackendConfiguration tests that the memory backend receives and utilizes the MaxQueueCapacity handler
+// configuration.
+//
+// This test works by enqueueing 3 jobs. Each job sleeps for longer than the test is willing to wait for the jobs to
+// enqueue. The `done` channel is notified when all 3 jobs are enqueued.
+//
+// By serializing handler execution with `WithOption(HandlerConcurrency(1))` and enqueueing jobs asynchronously, we can wait
+// on `done` and a timeout channel to see which one completes first.
+//
+// Since the queue has a capacity of 1 and the handler is serialized, we know that `done` cannot be notified until job 1
+// is complete, job 2 is processing, and job 3 can be added to the queue.
+//
+// If `done` is notified before the timeout channel, this test would fail, because that would mean Enqueue() is not
+// blocking while the first Sleep()ing job is running. If the qeueue is blocking when it meets its capacity, we know
+// that the max queue capacity configuration has taken effect.
+func TestMemeoryBackendConfiguration(t *testing.T) {
+	numJobs := 3
+	queue := "testing"
+	timeout := false
+
+	done := make(chan bool)
+
+	ctx := context.TODO()
+	backend, err := NewMemBackend()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	nq, err := New(ctx, Backend(backend))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	handler := NewHandler(func(ctx context.Context) (err error) {
+		time.Sleep(100 * time.Millisecond)
+		return
+	})
+
+	handler = handler.
+		WithOption(HandlerConcurrency(1)).
+		WithOption(MaxQueueCapacity(1))
+
+	nq.Listen(ctx, queue, handler)
+
+	go func() {
+		for i := 0; i < numJobs; i++ {
+			jid, err := nq.Enqueue(ctx, Job{
+				Queue: queue,
+				Payload: map[string]interface{}{
+					"message": fmt.Sprintf("hello world: %d", i),
+				},
+			})
+			if err != nil || jid == DuplicateJobID {
+				slog.Error("job was not enqueued. either it was duplicate or this error caused it:", err)
+			}
+		}
+
+		done <- true
+	}()
+
+	select {
+	case <-time.After(time.Duration(50) * time.Millisecond):
+		timeout = true
+	case <-done:
+		err = errors.New("this job should have timed out")
+	}
+
+	if !timeout {
+		t.Error(err)
+	}
+
+	nq.Shutdown(ctx)
+}
+
+func TestMemeoryBackendFutureJobScheduling(t *testing.T) {
+	queue := "testing"
+
+	ctx := context.TODO()
+	backend, err := NewMemBackend()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	nq, err := New(ctx, Backend(backend))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	handler := NewHandler(func(ctx context.Context) (err error) {
+		return
+	})
+
+	nq.Listen(ctx, queue, handler)
+
+	jid, err := nq.Enqueue(ctx, Job{
+		Queue: queue,
+		Payload: map[string]interface{}{
+			"message": "hello world",
+		},
+		RunAfter: time.Now().Add(5 * time.Second),
+	})
+	if err != nil || jid == DuplicateJobID {
+		slog.Error("job was not enqueued. either it was duplicate or this error caused it:", err)
+	}
+
+	mb := nq.(*MemBackend)
+
+	var ok bool
+	if _, ok = mb.futureJobs.Load(jid); !ok {
+		t.Error(err)
+	}
+
+	nq.Shutdown(ctx)
+}

neoq.go

@@ -2,9 +2,7 @@
 //
 // Neoq's goal is to minimize the infrastructure necessary to add background job processing to Go applications. It does so by implementing queue durability with modular backends, rather than introducing a strict dependency on a particular backend such as Redis.
 //
-// A Postgres backend is provided out of a box. However, for Neoq to meet its goal of reducing the infrastructure
-// necessary to run background jobs -- additional backends are necessary. E.g. Applications that use MySQL, MonogoDB, or
-// Redis as their primary data stores will ideally use Neoq with corresponding backends.
+// An in-memory and Postgres backend are provided out of the box.
 package neoq
 
 // TODO dependencies to factor out
@@ -12,6 +10,11 @@ package neoq
 // "github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
 import (
 	"context"
+	"crypto/md5"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
 	"math"
 	"runtime"
 	"time"
@@ -20,7 +23,6 @@ import (
 
 	"github.com/guregu/null"
 	"github.com/jackc/pgx/v5"
-	"github.com/pkg/errors"
 )
 
 type contextKey int
@@ -32,11 +34,10 @@ const (
 	JobStatusProcessed = "processed"
 	JobStatusFailed    = "failed"
 
-	DefaultPgConnectionString = "postgres://postgres:postgres@127.0.0.1:5432/neoq"
 	DefaultTransactionTimeout = time.Minute
 	DefaultHandlerDeadline    = 30 * time.Second
 	DuplicateJobID            = -1
-	postgresBackendName       = "postgres"
+	UnqueuedJobID             = -2
 )
 
 // Neoq interface is Neoq's primary API
@@ -53,10 +54,10 @@ type Neoq interface {
 	ListenCron(ctx context.Context, cron string, h Handler) (err error)
 
 	// Shutdown halts the worker
-	Shutdown(ctx context.Context) error
+	Shutdown(ctx context.Context) (err error)
 
 	// WithConfig configures neoq with with optional configuration
-	WithConfig(opt ConfigOption) Neoq
+	WithConfig(opt ConfigOption) (n Neoq)
 }
 
 // Logger interface is the interface that neoq's logger must implement
@@ -64,7 +65,7 @@ type Neoq interface {
 // This interface is a subset of [slog.Logger]. The slog interface was chosen under the assumption that its
 // likely to be Golang's standard library logging interface.
 //
-// TODO: Add WithLoggerOpt() for user-supplied logger configuration
+// TODO: Add WithLogger for user-supplied logger configuration
 type Logger interface {
 	Debug(msg string, args ...any)
 	Error(msg string, err error, args ...any)
@@ -76,9 +77,10 @@ type HandlerFunc func(ctx context.Context) error
 
 // Handler handles jobs on a queue
 type Handler struct {
-	handle      HandlerFunc
-	concurrency int
-	deadline    time.Duration
+	handle        HandlerFunc
+	concurrency   int
+	deadline      time.Duration
+	queueCapacity int64
 }
 
 // HandlerOption is function that sets optional configuration for Handlers
@@ -99,14 +101,22 @@ func HandlerDeadline(d time.Duration) HandlerOption {
 	}
 }
 
-// HandlerConcurreny configures Neoq handlers to process jobs concurrently
+// HandlerConcurrency configures Neoq handlers to process jobs concurrently
 // the default concurrency is the number of (v)CPUs on the machine running Neoq
-func HandlerConcurreny(c int) HandlerOption {
+func HandlerConcurrency(c int) HandlerOption {
 	return func(h *Handler) {
 		h.concurrency = c
 	}
 }
 
+// MaxQueueCapacity configures Handlers to enforce a maximum capacity on the queues that it handles
+// queues that have reached capacity cause Enqueue() to block until the queue is below capacity
+func MaxQueueCapacity(capacity int64) HandlerOption {
+	return func(h *Handler) {
+		h.queueCapacity = capacity
+	}
+}
+
 // NewHandler creates a new queue handler
 func NewHandler(f HandlerFunc, opts ...HandlerOption) (h Handler) {
 	h = Handler{
@@ -134,6 +144,7 @@ type ConfigOption func(n Neoq)
 // Jobs are what are placed on queues for processing.
 //
 // The Fingerprint field can be supplied by the user to impact job deduplication.
+// TODO: Factor out `null` usage
 type Job struct {
 	ID          int64          `db:"id"`
 	Fingerprint string         `db:"fingerprint"` // A md5 sum of the job's queue + payload, affects job deduplication
@@ -167,12 +178,8 @@ func New(ctx context.Context, opts ...ConfigOption) (n Neoq, err error) {
 			return
 		}
 		n, err = NewPgBackend(ctx, ic.connectionString, opts...)
-	// TODO make the default an in-memory backend
 	default:
-		if ic.connectionString == "" {
-			ic.connectionString = DefaultPgConnectionString
-		}
-		n, err = NewPgBackend(ctx, ic.connectionString, opts...)
+		n, err = NewMemBackend(opts...)
 	}
 
 	return
@@ -213,7 +220,7 @@ func Backend(backend Neoq) ConfigOption {
 	}
 }
 
-// ConnectionString is a configuration option that sets a connection string for backend initialization:w
+// ConnectionString is a configuration option that sets a connection string for backend initialization
 func ConnectionString(connectionString string) ConfigOption {
 	return func(n Neoq) {
 		switch c := n.(type) {
@@ -256,8 +263,8 @@ func calculateBackoff(retryCount int) time.Time {
 }
 
 func randInt(max int) int {
-	rand.Seed(time.Now().UnixNano())
-	return rand.Intn(max)
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	return r.Intn(max)
 }
 
 // internalConfig models internal neoq configuratio not exposed to users
@@ -286,3 +293,47 @@ func (i internalConfig) Shutdown(ctx context.Context) (err error) {
 func (i internalConfig) WithConfig(opt ConfigOption) (n Neoq) {
 	return
 }
+
+// exechandler executes handler functions with a concrete time deadline
+func execHandler(ctx context.Context, handler Handler) (err error) {
+	deadlineCtx, cancel := context.WithDeadline(ctx, time.Now().Add(handler.deadline))
+	defer cancel()
+
+	var errCh = make(chan error, 1)
+	var done = make(chan bool)
+	go func(ctx context.Context) (e error) {
+		errCh <- handler.handle(ctx)
+		done <- true
+		return
+	}(ctx)
+
+	select {
+	case <-done:
+		err = <-errCh
+		return
+	case <-deadlineCtx.Done():
+		err = fmt.Errorf("job exceeded its %s deadline", handler.deadline)
+	}
+
+	return
+}
+
+// fingerprintJob fingerprints jobs as an md5 hash of its queue combined with its JSON-serialized payload
+func fingerprintJob(j *Job) (err error) {
+	// only generate a fingerprint if the job is not already fingerprinted
+	if j.Fingerprint != "" {
+		return
+	}
+
+	var js []byte
+	js, err = json.Marshal(j.Payload)
+	if err != nil {
+		return
+	}
+	h := md5.New()
+	io.WriteString(h, j.Queue)
+	io.WriteString(h, string(js))
+	j.Fingerprint = fmt.Sprintf("%x", h.Sum(nil))
+
+	return
+}

neoq_test.go

@@ -4,48 +4,36 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log"
-	"os"
 	"testing"
 	"time"
 )
 
-var (
-	dbURL = "postgres://postgres:postgres@127.0.0.1:5432/neoq"
-)
-
 func TestWorkerListenConn(t *testing.T) {
-	const queue = "foobar"
-	ctx := context.TODO()
-
-	cnx := os.Getenv("DATABASE_URL")
-	if cnx == "" {
-		cnx = dbURL
-	}
-
-	pgBackend, err := NewPgBackend(ctx, cnx)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	nq, err := New(ctx, Backend(pgBackend))
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	jobRan := false
+	const queue = "testing"
+	timeout := false
+	complete := false
 	numJobs := 1
+	doneCnt := 0
 	var done = make(chan bool, numJobs)
+
+	ctx := context.TODO()
+	backend, err := NewMemBackend()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	nq, err := New(ctx, Backend(backend))
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	handler := NewHandler(func(ctx context.Context) (err error) {
-		var j *Job
-		j, err = JobFromContext(ctx)
-		log.Println("queue:", j.Queue, "got job", "id:", j.ID, "messsage:", j.Payload["message"])
 		done <- true
 		return
 	})
 	handler = handler.
 		WithOption(HandlerDeadline(500 * time.Millisecond)).
-		WithOption(HandlerConcurreny(1))
+		WithOption(HandlerConcurrency(1))
 
 	if err != nil {
 		t.Error(err)
@@ -55,7 +43,7 @@ func TestWorkerListenConn(t *testing.T) {
 	nq.Listen(ctx, queue, handler)
 
 	// allow time for listener to start
-	time.Sleep(50 * time.Millisecond)
+	time.Sleep(5 * time.Millisecond)
 
 	for i := 0; i < numJobs; i++ {
 		jid, err := nq.Enqueue(ctx, Job{
@@ -69,8 +57,6 @@ func TestWorkerListenConn(t *testing.T) {
 		}
 	}
 
-	timeout := false
-	doneCnt := 0
 	for {
 		select {
 		case <-time.After(5 * time.Second):
@@ -81,7 +67,7 @@ func TestWorkerListenConn(t *testing.T) {
 		}
 
 		if doneCnt >= numJobs {
-			jobRan = true
+			complete = true
 			break
 		}
 
@@ -90,10 +76,7 @@ func TestWorkerListenConn(t *testing.T) {
 		}
 	}
 
-	// Allow time for job status to be updated in the database
-	time.Sleep(50 * time.Millisecond)
-
-	if !jobRan {
+	if !complete {
 		t.Error(err)
 	}
 
@@ -107,17 +90,16 @@ func TestWorkerListenCron(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	jobRan := false
+	complete := false
 	var done = make(chan bool)
 	handler := NewHandler(func(ctx context.Context) (err error) {
-		log.Println("got periodic job")
 		done <- true
 		return
 	})
 
 	handler = handler.
 		WithOption(HandlerDeadline(500 * time.Millisecond)).
-		WithOption(HandlerConcurreny(1))
+		WithOption(HandlerConcurrency(1))
 
 	if err != nil {
 		t.Error(err)
@@ -126,19 +108,16 @@ func TestWorkerListenCron(t *testing.T) {
 	nq.ListenCron(ctx, cron, handler)
 
 	// allow time for listener to start
-	time.Sleep(50 * time.Millisecond)
+	time.Sleep(5 * time.Millisecond)
 
 	select {
 	case <-time.After(5 * time.Second):
 		err = errors.New("timed out waiting for periodic job")
 	case <-done:
-		jobRan = true
+		complete = true
 	}
 
-	// Allow time for job status to be updated in the database
-	time.Sleep(50 * time.Millisecond)
-
-	if !jobRan {
+	if !complete {
 		t.Error(err)
 	}
 }

postgres_backend.go

@@ -2,10 +2,8 @@ package neoq
 
 import (
 	"context"
-	"crypto/md5"
-	"encoding/json"
+	"errors"
 	"fmt"
-	"io"
 	"log"
 	"os"
 	"strconv"
@@ -17,13 +15,14 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
-	"github.com/pkg/errors"
 	"github.com/robfig/cron"
 	"golang.org/x/exp/slog"
 )
 
 const (
-	PendingJobIDQuery = `SELECT id
+	postgresBackendName       = "postgres"
+	DefaultPgConnectionString = "postgres://postgres:postgres@127.0.0.1:5432/neoq"
+	PendingJobIDQuery         = `SELECT id
 					FROM neoq_jobs
 					WHERE queue = $1
 					AND status NOT IN ('processed')
@@ -59,6 +58,7 @@ type PgBackend struct {
 	logger     Logger
 }
 
+// NewPgBackend creates a new neoq backend backed by Postgres
 // If the database does not yet exist, Neoq will attempt to create the database and related tables by default.
 //
 // Connection strings may be a URL or DSN-style connection string to neoq's database. The connection string supports multiple
@@ -388,18 +388,9 @@ func (w PgBackend) WithConfig(opt ConfigOption) Neoq {
 // Duplicate jobs are not added to the queue. Any two unprocessed jobs with the same fingerprint are duplicates
 func (w PgBackend) enqueueJob(ctx context.Context, tx pgx.Tx, j Job) (jobID int64, err error) {
 
-	// fingerprint the job if it hasn't been fingerprinted already
-	// a fingerprint is the md5 sum of: queue + payload
-	if j.Fingerprint == "" {
-		var js []byte
-		js, err = json.Marshal(j.Payload)
-		if err != nil {
-			return
-		}
-		h := md5.New()
-		io.WriteString(h, j.Queue)
-		io.WriteString(h, string(js))
-		j.Fingerprint = fmt.Sprintf("%x", h.Sum(nil))
+	err = fingerprintJob(&j)
+	if err != nil {
+		return
 	}
 
 	var rowCount int64
@@ -525,6 +516,7 @@ func (w PgBackend) start(ctx context.Context, queue string) (err error) {
 
 	for i := 0; i < handler.concurrency; i++ {
 		go func() {
+			var err error
 			var jobID int64
 
 			for {
@@ -710,38 +702,17 @@ func (w PgBackend) handleJob(ctx context.Context, jobID int64, handler Handler)
 	}
 
 	// execute the queue handler of this job
-	handlerErr := w.execHandler(ctx, handler)
+	handlerErr := execHandler(ctx, handler)
 	err = w.updateJob(ctx, handlerErr)
 	if err != nil {
-		err = errors.Wrap(err, "error updating job status")
+		err = fmt.Errorf("error updating job status: %w", err)
 		return
 	}
 
 	err = tx.Commit(ctx)
 	if err != nil {
 		w.logger.Error("unable to commit job transaction. retrying this job may dupliate work", err, "job_id", job.ID)
-		err = errors.Wrap(err, "unable to commit job transaction. retrying this job may dupliate work")
-	}
-
-	return
-}
-
-// exechandler executes handler functions with a concrete time deadline
-func (w PgBackend) execHandler(ctx context.Context, handler Handler) (err error) {
-	deadlineCtx, cancel := context.WithDeadline(ctx, time.Now().Add(handler.deadline))
-	defer cancel()
-
-	var done = make(chan bool)
-	go func(ctx context.Context) {
-		err = handler.handle(ctx)
-		done <- true
-	}(ctx)
-
-	select {
-	case <-done:
-		return
-	case <-deadlineCtx.Done():
-		err = fmt.Errorf("job exceeded its %s deadline", handler.deadline)
+		err = fmt.Errorf("unable to commit job transaction. retrying this job may dupliate work: %w", err)
 	}
 
 	return
@@ -757,9 +728,8 @@ func (w PgBackend) listen(ctx context.Context, queue string) (c chan int64) {
 	// set this connection's idle in transaction timeout to infinite so it is not intermittently disconnected
 	_, err = w.listenConn.Exec(ctx, fmt.Sprintf("SET idle_in_transaction_session_timeout = '0'; LISTEN %s", queue))
 	if err != nil {
-		msg := "unable to create database connection for listener"
-		err = errors.Wrap(err, msg)
-		w.logger.Error(msg, err)
+		err = fmt.Errorf("unable to create database connection for listener: %w", err)
+		w.logger.Error("unablet o create database connection for listener", err)
 		return
 	}