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