fix: unprocessed jobs from unwatched queues block watched queues

There could be a condition where jobs were added to a queue that is not longer being watched,
yet the jobs in those queues would continue appearing in the pending job list because
the query to get pending jobs did not filter by queues being watched.

Author: acaloiaro

backends/postgres/postgres_backend.go

@@ -5,8 +5,12 @@ import (
 	"embed"
 	"errors"
 	"fmt"
+	"log/slog"
+	"maps"
 	"net/url"
 	"os"
+	"slices"
+	"strings"
 	"sync"
 	"time"
 
@@ -26,8 +30,6 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
 	"github.com/robfig/cron"
-	"golang.org/x/exp/slices"
-	"golang.org/x/exp/slog"
 )
 
 //go:embed migrations/*.sql
@@ -46,6 +48,7 @@ const (
 	PendingJobsQuery = `SELECT id,fingerprint,queue,status,deadline,payload,retries,max_retries,run_after,ran_at,created_at,error
 					FROM neoq_jobs
 					WHERE status NOT IN ('processed')
+					AND queue IN ($1)
 					AND run_after <= NOW()
 					ORDER BY created_at ASC
 					FOR UPDATE SKIP LOCKED
@@ -220,16 +223,13 @@ func (p *PgBackend) listenerManager(ctx context.Context) {
 			p.listenerConnMu.Lock()
 			p.listenerConn = lc
 			p.mu.Lock()
-			handlers := p.handlers
-			p.mu.Unlock()
-
-			for queue := range handlers {
+			for queue := range p.handlers {
 				_, err = p.listenerConn.Exec(ctx, fmt.Sprintf(`LISTEN %q`, queue))
 				if err != nil {
 					p.logger.Error("unable to listen on queue", slog.Any("error", err), slog.String("queue", queue))
 				}
 			}
-
+			p.mu.Unlock()
 			p.listenerConnMu.Unlock()
 
 			p.logger.Debug("worker database connection established")
@@ -835,14 +835,22 @@ func (p *PgBackend) announceJob(ctx context.Context, queue, jobID string) {
 
 // processPendingJobs starts a goroutine that periodically fetches pendings jobs and announces them to workers.
 //
+// The interval for this task does not need to be particularly short unless an application suffers frequent database
+// disconnects, as most pending jobs are picked up by LISTEN after being announced with NOTIFY.
+//
 // Past due jobs are fetched on the interval [neoq.DefaultPendingJobFetchInterval]
 // nolint: cyclop
 func (p *PgBackend) processPendingJobs(ctx context.Context) {
 	go func(ctx context.Context) {
 		var err error
 		var conn *pgxpool.Conn
 		var pendingJobs []*jobs.Job
-		ticker := time.NewTicker(neoq.DefaultPendingJobFetchInterval)
+		var ticker *time.Ticker
+		if p.config.PendingJobCheckInterval > 0 {
+			ticker = time.NewTicker(p.config.PendingJobCheckInterval)
+		} else {
+			ticker = time.NewTicker(neoq.DefaultPendingJobFetchInterval)
+		}
 
 		// check for pending jobs on an interval until the context is canceled
 		for {
@@ -1044,7 +1052,11 @@ func (p *PgBackend) getJob(ctx context.Context, tx pgx.Tx, jobID string) (job *j
 }
 
 func (p *PgBackend) getPendingJobs(ctx context.Context, conn *pgxpool.Conn) (pendingJobs []*jobs.Job, err error) {
-	rows, err := conn.Query(ctx, PendingJobsQuery)
+	p.mu.Lock()
+	// convert watched queue map to string: "'queue1', 'queue2', ..." for use in Postgres IN statement
+	activeQueues := slices.Collect(maps.Keys(p.handlers))
+	p.mu.Unlock()
+	rows, err := conn.Query(ctx, PendingJobsQuery, strings.Join(activeQueues, ","))
 	if err != nil {
 		return
 	}

backends/postgres/postgres_backend_test.go

@@ -1144,26 +1144,27 @@ func TestProcessPendingJobs(t *testing.T) {
 
 	ctx := context.Background()
 
-	// INSERTing jobs into the job queue before noeq is listening on any queues ensures that the new job is not announced, and when
+	nq, err := neoq.New(ctx, neoq.WithBackend(postgres.Backend), postgres.WithConnectionString(connString), neoq.WithPendingJobCheckInterval(time.Duration(50*time.Millisecond)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nq.Shutdown(ctx)
+
+	// Adding jobs into the job queue before noeq is listening on any queues ensures that the new job is not announced, and when
 	// neoq _is_ started, that there is a pending jobs waiting to be processed
 	payload := map[string]interface{}{
 		"message": "hello world",
 	}
 	var pendingJobID string
-	err := conn.QueryRow(ctx, `INSERT INTO neoq_jobs(queue, fingerprint, payload, run_after, deadline, max_retries)
-		VALUES ($1, $2, $3, $4, $5, $6) RETURNING id`,
-		queue, "dummy", payload, time.Now().UTC(), nil, 1).Scan(&pendingJobID)
+	pendingJobID, err = nq.Enqueue(ctx, &jobs.Job{
+		Queue:   queue,
+		Payload: payload,
+	})
 	if err != nil {
 		t.Error(fmt.Errorf("unable to add job to queue: %w", err))
 		return
 	}
 
-	nq, err := neoq.New(ctx, neoq.WithBackend(postgres.Backend), postgres.WithConnectionString(connString))
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer nq.Shutdown(ctx)
-
 	h := handler.New(queue, func(_ context.Context) (err error) {
 		return
 	})

go.mod

@@ -1,6 +1,6 @@
 module github.com/acaloiaro/neoq
 
-go 1.21
+go 1.24.0
 
 require (
 	github.com/golang-migrate/migrate/v4 v4.16.2

neoq.go

@@ -18,7 +18,7 @@ const (
 	DefaultPendingJobFetchInterval = 60 * time.Second
 	// the window of time between time.Now() and when a job's RunAfter comes due that neoq will schedule a goroutine to
 	// schdule the job for execution.
-	// E.g. right now is 16:00 and a job's RunAfter is 16:30 of the same date. This job will get a dedicated goroutine
+	// E.g. right now is 16:00:00 and a job's RunAfter is 16:00:30 of the same date. This job will get a dedicated goroutine
 	// to wait until the job's RunAfter, scheduling the job to be run exactly at RunAfter
 	DefaultFutureJobWindow  = 30 * time.Second
 	DefaultJobCheckInterval = 1 * time.Second
@@ -32,18 +32,19 @@ var ErrBackendNotSpecified = errors.New("a backend must be specified")
 // backends. [BackendConcurrency], for example, is only used by the redis backend. Other backends manage concurrency on a
 // per-handler basis.
 type Config struct {
-	BackendInitializer     BackendInitializer
-	BackendAuthPassword    string                   // password with which to authenticate to the backend
-	BackendConcurrency     int                      // total number of backend processes available to process jobs
-	ConnectionString       string                   // a string containing connection details for the backend
-	JobCheckInterval       time.Duration            // the interval of time between checking for new future/past-due jobs
-	FutureJobWindow        time.Duration            // time duration between current time and job.RunAfter that future jobs get scheduled
-	IdleTransactionTimeout int                      // number of milliseconds PgBackend transaction may idle before the connection is killed
-	ShutdownTimeout        time.Duration            // duration to wait for jobs to finish during shutdown
-	SynchronousCommit      bool                     // Postgres: Enable synchronous commits (increases durability, decreases performance)
-	LogLevel               logging.LogLevel         // the log level of the default logger
-	PGConnectionTimeout    time.Duration            // the amount of time to wait for a connection to become available before timing out
-	RecoveryCallback       handler.RecoveryCallback // the recovery handler applied to all Handlers excuted by the associated Neoq instance
+	BackendInitializer      BackendInitializer
+	BackendAuthPassword     string                   // password with which to authenticate to the backend
+	BackendConcurrency      int                      // total number of backend processes available to process jobs
+	ConnectionString        string                   // a string containing connection details for the backend
+	JobCheckInterval        time.Duration            // the interval of time between checking for new future jobs
+	PendingJobCheckInterval time.Duration            // duration of time between checking for unprocessed jobs due to downtime/db disconnects
+	FutureJobWindow         time.Duration            // time duration between current time and job.RunAfter that future jobs get scheduled
+	IdleTransactionTimeout  int                      // number of milliseconds PgBackend transaction may idle before the connection is killed
+	ShutdownTimeout         time.Duration            // duration to wait for jobs to finish during shutdown
+	SynchronousCommit       bool                     // Postgres: Enable synchronous commits (increases durability, decreases performance)
+	LogLevel                logging.LogLevel         // the log level of the default logger
+	PGConnectionTimeout     time.Duration            // the amount of time to wait for a connection to become available before timing out
+	RecoveryCallback        handler.RecoveryCallback // the recovery handler applied to all Handlers excuted by the associated Neoq instance
 }
 
 // ConfigOption is a function that sets optional backend configuration
@@ -139,6 +140,17 @@ func WithJobCheckInterval(interval time.Duration) ConfigOption {
 	}
 }
 
+// WithPendingJobCheckInterval configures the duration of time between checking for unprocessed jobs on watched queues.
+//
+// Pending jobs are jobs that were queued while neoq wasn't listening or while a disconnect occurred. Most jobs are
+// processed via Postgres' LISTEN/NOTIFY facilities. This check interval is a last resort and does not need to be short.
+// Be careful not to hammer your database by setting this too low.
+func WithPendingJobCheckInterval(interval time.Duration) ConfigOption {
+	return func(c *Config) {
+		c.PendingJobCheckInterval = interval
+	}
+}
+
 // WithLogLevel configures the log level for neoq's default logger. By default, log level is "INFO".
 // if SetLogger is used, WithLogLevel has no effect on the set logger
 func WithLogLevel(level logging.LogLevel) ConfigOption {