Support optional override argument when enqueuing job

.github/workflows/golangci-lint.yml

@@ -12,10 +12,10 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/setup-go@v3
+      - uses: actions/setup-go@v5
         with:
           go-version: '1.21'
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: go mod
         run: make mod
       - name: golangci-lint

.github/workflows/goreleaser.yml

@@ -14,7 +14,7 @@ jobs:
     steps:
       -
         name: Checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           fetch-depth: 0
       -
@@ -26,7 +26,7 @@ jobs:
           DEFAULT_BUMP: minor
       -
         name: Set up Go
-        uses: actions/setup-go@v3
+        uses: actions/setup-go@v5
       -
         name: Run GoReleaser
         uses: goreleaser/goreleaser-action@v4

.github/workflows/test.yml

@@ -37,7 +37,7 @@ jobs:
     container: golang:1.21
 
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: test
         run: make mod test coverage
         env:

.github/workflows/todo_to_issue.yml

@@ -16,7 +16,7 @@ jobs:
     build:
         runs-on: "ubuntu-latest"
         steps:
-            - uses: "actions/checkout@v3"
+            - uses: "actions/checkout@v4"
             - name: "TODO to Issue"
               uses: "alstr/todo-to-issue-action@v4"
               env:

backends/memory/memory_backend.go

@@ -67,7 +67,12 @@ func Backend(_ context.Context, opts ...neoq.ConfigOption) (backend neoq.Neoq, e
 }
 
 // Enqueue queues jobs to be executed asynchronously
-func (m *MemBackend) Enqueue(_ context.Context, job *jobs.Job) (jobID string, err error) {
+func (m *MemBackend) Enqueue(_ context.Context, job *jobs.Job, jobOptions ...neoq.JobOption) (jobID string, err error) {
+	options := neoq.JobOptions{}
+	for _, opt := range jobOptions {
+		opt(&options)
+	}
+
 	var queueChan chan *jobs.Job
 	var qc any
 	var ok bool
@@ -96,29 +101,34 @@ func (m *MemBackend) Enqueue(_ context.Context, job *jobs.Job) (jobID string, er
 		job.RunAfter = now
 	}
 
-	if job.Queue == "" {
-		err = jobs.ErrNoQueueSpecified
-		return
-	}
-
 	err = jobs.FingerprintJob(job)
 	if err != nil {
+		err = errors.Join(jobs.ErrCantGenerateFingerprint, err)
 		return
 	}
 
 	// if the job fingerprint is already known, don't queue the job
 	if _, found := m.fingerprints.Load(job.Fingerprint); found {
-		return jobs.DuplicateJobID, nil
+		if !options.Override {
+			return jobs.DuplicateJobID, jobs.ErrJobFingerprintConflict
+		}
+		oldJob, found := m.fingerprints.Swap(job.Fingerprint, job)
+		if found {
+			// Return the same JobID to make it the same as posgres
+			job.ID = oldJob.(*jobs.Job).ID
+		} else {
+			m.logger.Info("Expected to get job but none was returned for fingerprint %s", job.Fingerprint)
+		}
+		jobID = fmt.Sprint(job.ID)
+	} else {
+		m.fingerprints.Store(job.Fingerprint, job)
+		m.mu.Lock()
+		m.jobCount++
+		m.mu.Unlock()
+		job.ID = m.jobCount
+		jobID = fmt.Sprint(m.jobCount)
 	}
 
-	m.fingerprints.Store(job.Fingerprint, job)
-	m.mu.Lock()
-	m.jobCount++
-	m.mu.Unlock()
-
-	job.ID = m.jobCount
-	jobID = fmt.Sprint(m.jobCount)
-
 	if job.RunAfter.Equal(now) {
 		queueChan <- job
 	} else {
@@ -135,6 +145,8 @@ func (m *MemBackend) Start(ctx context.Context, h handler.Handler) (err error) {
 		queueCapacity = defaultMemQueueCapacity
 	}
 
+	h.RecoverCallback = m.config.RecoveryCallback
+
 	m.handlers.Store(h.Queue, h)
 	m.queues.Store(h.Queue, make(chan *jobs.Job, queueCapacity))
 
@@ -173,6 +185,7 @@ func (m *MemBackend) StartCron(ctx context.Context, cronSpec string, h handler.H
 	m.cancelFuncs = append(m.cancelFuncs, cancel)
 	m.mu.Unlock()
 	h.Queue = queue
+	h.RecoverCallback = m.config.RecoveryCallback
 
 	err = m.Start(ctx, h)
 	if err != nil {
@@ -249,7 +262,7 @@ func (m *MemBackend) start(ctx context.Context, queue string) (err error) {
 
 					m.logger.Error("job failed", slog.Int64("job_id", job.ID), slog.Any("error", err))
 
-					runAfter := internal.CalculateBackoff(job.Retries)
+					runAfter := internal.CalculateBackoff(job.Retries, job.RunAfter)
 					job.RunAfter = runAfter
 					job.Status = internal.JobStatusFailed
 					m.queueFutureJob(job)
@@ -318,7 +331,7 @@ func (m *MemBackend) scheduleFutureJobs(ctx context.Context) {
 }
 
 func (m *MemBackend) handleJob(ctx context.Context, job *jobs.Job, h handler.Handler) (err error) {
-	ctx = withJobContext(ctx, job)
+	ctx = jobs.WithJobContext(ctx, job)
 
 	m.logger.Debug(
 		"handling job",
@@ -378,8 +391,3 @@ func (m *MemBackend) removeFutureJob(jobID int64) {
 		m.futureJobs.Delete(job.ID)
 	}
 }
-
-// withJobContext creates a new context with the Job set
-func withJobContext(ctx context.Context, j *jobs.Job) context.Context {
-	return context.WithValue(ctx, internal.JobCtxVarKey, j)
-}

backends/memory/memory_backend_test.go

@@ -11,19 +11,21 @@ import (
 	"time"
 
 	"github.com/acaloiaro/neoq"
+	"github.com/acaloiaro/neoq/backends"
 	"github.com/acaloiaro/neoq/backends/memory"
 	"github.com/acaloiaro/neoq/handler"
 	"github.com/acaloiaro/neoq/jobs"
 	"github.com/acaloiaro/neoq/logging"
 	"github.com/acaloiaro/neoq/testutils"
 	"github.com/pkg/errors"
 	"github.com/robfig/cron"
+	"github.com/stretchr/testify/suite"
 	"golang.org/x/exp/slog"
 )
 
-const (
-	queue = "testing"
-)
+var queue = "testing"
+var q1 = "queue1"
+var q2 = "queue2"
 
 var (
 	errPeriodicTimeout = errors.New("timed out waiting for periodic job")
@@ -231,9 +233,6 @@ func TestFutureJobSchedulingMultipleQueues(t *testing.T) {
 	jobsProcessed1 := 0
 	jobsProcessed2 := 0
 
-	q1 := "queue1"
-	q2 := "queue2"
-
 	done1 := make(chan bool)
 	done2 := make(chan bool)
 
@@ -427,3 +426,17 @@ result_loop:
 		t.Error(err)
 	}
 }
+
+func initQueue() (neoq.Neoq, error) {
+	ctx := context.Background()
+	return neoq.New(ctx, neoq.WithBackend(memory.Backend), neoq.WithLogLevel(logging.LogLevelDebug)) //nolint: error
+}
+
+func TestSuite(t *testing.T) {
+	n, err := initQueue()
+	if err != nil {
+		t.Fatal(err)
+	}
+	s := backends.NewNeoQTestSuite(n)
+	suite.Run(t, s)
+}

backends/postgres/migrations/20240302172413_make_fingerprints_unique_to_each_queue.up.sql

@@ -0,0 +1,2 @@
+DROP INDEX IF EXISTS neoq_jobs_fingerprint_unique_idx;
+CREATE UNIQUE INDEX IF NOT EXISTS neoq_jobs_fingerprint_unique_idx ON neoq_jobs (queue, fingerprint, status) WHERE NOT (status = 'processed');