fix: execessive connection usage when monitoring pending jobs

The previous release attempted to fix an issue where, when LISTEN connections
drop an reconnect, some job announcements might be missed.

This fix reduces the number of connections required to do so.

backends/postgres/postgres_backend.go

@@ -43,10 +43,9 @@ const (
 					AND status NOT IN ('processed')
 					FOR UPDATE SKIP LOCKED
 					LIMIT 1`
-	PendingJobIDsQuery = `SELECT id
+	PendingJobsQuery = `SELECT id,fingerprint,queue,status,deadline,payload,retries,max_retries,run_after,ran_at,created_at,error
 					FROM neoq_jobs
-					WHERE queue = $1
-					AND status NOT IN ('processed')
+					WHERE status NOT IN ('processed')
 					AND run_after <= NOW()
 					ORDER BY created_at ASC
 					FOR UPDATE SKIP LOCKED
@@ -191,6 +190,13 @@ func Backend(ctx context.Context, opts ...neoq.ConfigOption) (pb neoq.Neoq, err
 	// monitor handlers for changes and LISTEN when new queues are added
 	go p.listenerManager(ctx)
 
+	// monitor queues for pending jobs, so neoq is resilient to LISTEN disconnects and reconnections
+	pendingJobsConn, err := p.pool.Acquire(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("unable to get a database connection: %w", err)
+	}
+	p.processPendingJobs(ctx, pendingJobsConn)
+
 	p.listenConnDown <- true
 
 	p.cron.Start()
@@ -200,7 +206,8 @@ func Backend(ctx context.Context, opts ...neoq.ConfigOption) (pb neoq.Neoq, err
 	return pb, nil
 }
 
-// listenerManager manages the LISTENer connection and adding queue to it
+// listenerManager manages the LISTENer connection and add queues to it
+// nolint: cyclop
 func (p *PgBackend) listenerManager(ctx context.Context) {
 	var err error
 	for {
@@ -675,7 +682,6 @@ func (p *PgBackend) start(ctx context.Context, h handler.Handler) (err error) {
 		return fmt.Errorf("%w: %s", handler.ErrNoHandlerForQueue, h.Queue)
 	}
 
-	pendingJobsChan := p.processPendingJobs(ctx, h.Queue)
 	// wait for the listener to connect and be ready to listen
 	for q := range p.readyQueues {
 		if q == h.Queue {
@@ -700,8 +706,6 @@ func (p *PgBackend) start(ctx context.Context, h handler.Handler) (err error) {
 				select {
 				case n = <-listenJobChan:
 					err = p.handleJob(ctx, n.Payload)
-				case n = <-pendingJobsChan:
-					err = p.handleJob(ctx, n.Payload)
 				case <-ctx.Done():
 					return
 				case <-errCh:
@@ -825,40 +829,29 @@ func (p *PgBackend) announceJob(ctx context.Context, queue, jobID string) {
 
 // processPendingJobs starts a goroutine that periodically fetches pendings jobs and announces them to workers.
 //
-// Past due jobs are fetched on the interval [neoq.Config.JobCheckInterval]
+// Past due jobs are fetched on the interval [neoq.DefaultPendingJobFetchInterval]
 // nolint: cyclop
-func (p *PgBackend) processPendingJobs(ctx context.Context, queue string) (jobsCh chan *pgconn.Notification) {
-	conn, err := p.acquire(ctx)
-	if err != nil {
-		p.logger.Error(
-			"failed to acquire database connection to listen for pending queue items",
-			slog.String("queue", queue),
-			slog.Any("error", err),
-		)
-		return
-	}
-
-	// check for new past-due jobs on an interval
-	ticker := time.NewTicker(p.config.JobCheckInterval)
+func (p *PgBackend) processPendingJobs(ctx context.Context, conn *pgxpool.Conn) {
 	go func(ctx context.Context) {
 		defer conn.Release()
+		ticker := time.NewTicker(neoq.DefaultPendingJobFetchInterval)
+
 		// check for pending jobs on an interval until the context is canceled
 		for {
-			jobIDs, err := p.getPendingJobIDs(ctx, conn, queue)
+			pendingJobs, err := p.getPendingJobs(ctx, conn)
 			if errors.Is(err, context.Canceled) {
 				return
 			}
 
 			if err != nil && !errors.Is(err, pgx.ErrNoRows) {
 				p.logger.Error(
-					"failed to fetch pending job",
-					slog.String("queue", queue),
+					"failed to fetch pending jobs",
 					slog.Any("error", err),
 				)
 			}
 
-			for _, jid := range jobIDs {
-				jobsCh <- &pgconn.Notification{Channel: queue, Payload: jid}
+			for _, job := range pendingJobs {
+				p.announceJob(ctx, job.Queue, fmt.Sprint(job.ID))
 			}
 			select {
 			case <-ctx.Done():
@@ -867,9 +860,6 @@ func (p *PgBackend) processPendingJobs(ctx context.Context, queue string) (jobsC
 			}
 		}
 	}(ctx)
-
-	jobsCh = make(chan *pgconn.Notification)
-	return jobsCh
 }
 
 // handleJob is the workhorse of Neoq
@@ -955,9 +945,6 @@ func (p *PgBackend) handleJob(ctx context.Context, jobID string) (err error) {
 }
 
 // listen uses Postgres LISTEN to listen for jobs on a queue
-// TODO: There is currently no handling of listener disconnects in PgBackend.
-// This will lead to jobs not getting processed until the worker is restarted.
-// Implement disconnect handling.
 func (p *PgBackend) listen(ctx context.Context) (c chan *pgconn.Notification, errCh chan error) {
 	c = make(chan *pgconn.Notification)
 	errCh = make(chan error)
@@ -1041,17 +1028,17 @@ func (p *PgBackend) getJob(ctx context.Context, tx pgx.Tx, jobID string) (job *j
 	return
 }
 
-func (p *PgBackend) getPendingJobIDs(ctx context.Context, conn *pgxpool.Conn, queue string) (jobIDs []string, err error) {
-	var rows pgx.Rows
-	var jid int64
-	rows, err = conn.Query(ctx, PendingJobIDsQuery, queue)
-	for rows.Next() {
-		err = rows.Scan(&jid)
+func (p *PgBackend) getPendingJobs(ctx context.Context, conn *pgxpool.Conn) (pendingJobs []*jobs.Job, err error) {
+	rows, err := conn.Query(ctx, PendingJobsQuery)
 	if err != nil {
 		return
 	}
-		jobIDs = append(jobIDs, fmt.Sprint(jid))
+
+	pendingJobs, err = pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[jobs.Job])
+	if err != nil {
+		return
 	}
+
 	return
 }
 

backends/postgres/postgres_backend_test.go

@@ -1132,3 +1132,74 @@ func TestHandlerRecoveryCallback(t *testing.T) {
 		t.Error(err)
 	}
 }
+
+// TestProcessPendingJobs tests that unanounced jobs with a run_after before the current timestamp get run periodically
+// This ensures that when LISTENER connections fails, that jobs that would have been announced still get processed eventually
+func TestProcessPendingJobs(t *testing.T) {
+	connString, conn := prepareAndCleanupDB(t)
+	const queue = "testing"
+	timeoutTimer := time.After(5 * time.Second)
+	done := make(chan bool)
+	defer close(done)
+
+	ctx := context.Background()
+
+	// INSERTing jobs into the 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)
+	if err != nil {
+		err = 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
+	})
+
+	// Start ensures that pending jobs will be processed
+	err = nq.Start(ctx, h)
+	if err != nil {
+		t.Error(err)
+	}
+
+	var status string
+	go func() {
+		// ensure job has failed/has the correct status
+		for {
+			err = conn.
+				QueryRow(context.Background(), "SELECT status FROM neoq_jobs WHERE id = $1", pendingJobID).
+				Scan(&status)
+			if err != nil {
+				break
+			}
+
+			if status != internal.JobStatusNew {
+				done <- true
+				break
+			}
+
+			time.Sleep(50 * time.Millisecond)
+		}
+	}()
+
+	select {
+	case <-timeoutTimer:
+		err = jobs.ErrJobTimeout
+	case <-done:
+	}
+	if err != nil {
+		t.Errorf("job should have resulted in a status of 'processed', but its status is %s", status)
+	}
+}

gomod2nix.toml

@@ -77,20 +77,20 @@ schema = 3
     version = "v1.24.0"
     hash = "sha256-yLzjFbMWnc5b033gcPLGP0KY1xWPJ3sjnUG/RndmC3o="
   [mod."golang.org/x/crypto"]
-    version = "v0.17.0"
-    hash = "sha256-/vzBaeD/Ymyc7cpjBvSfJfuZ57zWa9LOaZM7b33eIx0="
+    version = "v0.31.0"
+    hash = "sha256-ZBjoG7ZOuTEmjaXPP9txAvjAjC46DeaLs0zrNzi8EQw="
   [mod."golang.org/x/exp"]
     version = "v0.0.0-20230713183714-613f0c0eb8a1"
     hash = "sha256-VLE9CCOYpTdyBWaQ1YxXpGOBS74wpIR3JJ+JVzNyEkQ="
   [mod."golang.org/x/sync"]
-    version = "v0.2.0"
-    hash = "sha256-hKk9zsy2aXY7R0qGFZhGOVvk5qD17f6KHEuK4rGpTsg="
+    version = "v0.10.0"
+    hash = "sha256-HWruKClrdoBKVdxKCyoazxeQV4dIYLdkHekQvx275/o="
   [mod."golang.org/x/sys"]
-    version = "v0.15.0"
-    hash = "sha256-n7TlABF6179RzGq3gctPDKDPRtDfnwPdjNCMm8ps2KY="
+    version = "v0.28.0"
+    hash = "sha256-kzSlDo5FKsQU9cLefIt2dueGUfz9XuEW+mGSGlPATGc="
   [mod."golang.org/x/text"]
-    version = "v0.14.0"
-    hash = "sha256-yh3B0tom1RfzQBf1RNmfdNWF1PtiqxV41jW1GVS6JAg="
+    version = "v0.21.0"
+    hash = "sha256-QaMwddBRnoS2mv9Y86eVC2x2wx/GZ7kr2zAJvwDeCPc="
   [mod."golang.org/x/time"]
     version = "v0.0.0-20190308202827-9d24e82272b4"
     hash = "sha256-azbksMSLQf1CK0jF2i+ESjFenMDR88xRW1tov0metrg="

neoq.go

@@ -12,6 +12,10 @@ import (
 
 const (
 	DefaultIdleTxTimeout = 30000
+	// The duration of time between checking if any queues have pending jobs
+	// It's necessary to check for pending jobs periodically because if the LISTENer connection fails at any point, there is a period
+	// of time when new jobs get announced by the new job trigger announcement, but no listeners are LISTENing for work.
+	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