From 17c86fa00e8c22866db7a04efeed0381519d102b Mon Sep 17 00:00:00 2001 From: Adriano Caloiaro Date: Sat, 25 Jan 2025 17:08:46 -0700 Subject: [PATCH] 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 | 71 +++++++++------------- backends/postgres/postgres_backend_test.go | 71 ++++++++++++++++++++++ gomod2nix.toml | 16 ++--- neoq.go | 4 ++ 4 files changed, 112 insertions(+), 50 deletions(-) diff --git a/backends/postgres/postgres_backend.go b/backends/postgres/postgres_backend.go index 96ba00a..a16e546 100644 --- a/backends/postgres/postgres_backend.go +++ b/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) - if err != nil { - return - } - jobIDs = append(jobIDs, fmt.Sprint(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 } + + pendingJobs, err = pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[jobs.Job]) + if err != nil { + return + } + return } diff --git a/backends/postgres/postgres_backend_test.go b/backends/postgres/postgres_backend_test.go index 2198f16..d2a2ff3 100644 --- a/backends/postgres/postgres_backend_test.go +++ b/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) + } +} diff --git a/gomod2nix.toml b/gomod2nix.toml index ec2555f..f9ceda0 100644 --- a/gomod2nix.toml +++ b/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=" diff --git a/neoq.go b/neoq.go index 329a151..31d3dd6 100644 --- a/neoq.go +++ b/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