Bug fixing for postgres and memory backends, polish the unit tests

backends/memory/memory_backend.go

@@ -112,15 +112,23 @@ func (m *MemBackend) Enqueue(ctx context.Context, job *jobs.Job, jobOptions ...n
 		if !options.Override {
 			return jobs.DuplicateJobID, jobs.ErrJobFingerprintConflict
 		}
-	}
-	m.fingerprints.Store(job.Fingerprint, job)
-
-	m.mu.Lock()
-	m.jobCount++
-	m.mu.Unlock()
+		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)
 
-	job.ID = m.jobCount
-	jobID = fmt.Sprint(m.jobCount)
+	} else {
+		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

backends/memory/memory_backend_test.go

@@ -428,13 +428,11 @@ result_loop:
 	}
 }
 
-func TestFutureJobSchedulingOverride(t *testing.T) {
+func TestSuite(t *testing.T) {
 	ctx := context.Background()
-	nq, err := neoq.New(ctx, neoq.WithBackend(memory.Backend))
+	n, err := neoq.New(ctx, neoq.WithBackend(memory.Backend), neoq.WithLogLevel(logging.LogLevelDebug))
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer nq.Shutdown(ctx)
-
-	backends.TestOverrideFingerprint(t, ctx, nq)
+	backends.NewNeoQTestSuite(n).Run(t)
 }

backends/postgres/migrations/20240101191302_add_unique_on_fingerprint_constraint.down.sql

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

backends/postgres/migrations/20240101191302_add_unique_on_fingerprint_constraint.up.sql

@@ -1 +1,4 @@
-ALTER TABLE neoq_jobs ADD CONSTRAINT neoq_jobs_fingerprint_constraint_idx UNIQUE (queue, fingerprint, status, ran_at);
+DROP INDEX IF EXISTS neoq_jobs_fingerprint_idx;
+DROP INDEX IF EXISTS neoq_jobs_fingerprint_unique_idx;
+
+CREATE UNIQUE INDEX IF NOT EXISTS neoq_jobs_fingerprint_unique_idx ON neoq_jobs (queue, status, fingerprint, ran_at);

backends/postgres/postgres_backend.go

@@ -548,11 +548,14 @@ func (p *PgBackend) enqueueJob(ctx context.Context, tx pgx.Tx, j *jobs.Job, opti
 			j.Queue, j.Fingerprint, j.Payload, j.RunAfter, j.Deadline, j.MaxRetries).Scan(&jobID)
 	} else {
 		err = tx.QueryRow(ctx, `INSERT INTO neoq_jobs(queue, fingerprint, payload, run_after, deadline, max_retries)
-		VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (queue, fingerprint, status, ran_at) DO
+		VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (queue, status, fingerprint, ran_at) DO
 		UPDATE SET 
 		payload=$3, run_after=$4, deadline=$5, max_retries=$6		
 		RETURNING id`,
 			j.Queue, j.Fingerprint, j.Payload, j.RunAfter, j.Deadline, j.MaxRetries).Scan(&jobID)
+		if err != nil {
+			p.logger.Error("error enqueueing override job", slog.Any("error", err))
+		}
 	}
 
 	if err != nil {
@@ -849,19 +852,19 @@ func (p *PgBackend) handleJob(ctx context.Context, jobID string) (err error) {
 	var tx pgx.Tx
 	conn, err := p.acquire(ctx)
 	if err != nil {
-		return
+		return err
 	}
 	defer conn.Release()
 
 	tx, err = conn.Begin(ctx)
 	if err != nil {
-		return
+		return err
 	}
 	defer func(ctx context.Context) { _ = tx.Rollback(ctx) }(ctx) // rollback has no effect if the transaction has been committed
 
 	job, err = p.getJob(ctx, tx, jobID)
 	if err != nil {
-		return
+		return err
 	}
 
 	if job.Deadline != nil && job.Deadline.Before(time.Now().UTC()) {

backends/postgres/postgres_backend_test.go

@@ -777,15 +777,18 @@ func Test_ConnectionTimeout(t *testing.T) {
 		t.Error(err)
 	}
 }
-
-func TestFutureJobSchedulingOverride(t *testing.T) {
+func setup(t *testing.T) (neoq.Neoq, context.Context) {
 	connString, _ := prepareAndCleanupDB(t)
 
 	ctx := context.TODO()
 	nq, err := neoq.New(ctx, neoq.WithBackend(postgres.Backend), postgres.WithConnectionString(connString))
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer nq.Shutdown(ctx)
-	backends.TestOverrideFingerprint(t, ctx, nq)
+	return nq, ctx
+}
+
+func TestSuite(t *testing.T) {
+	n, _ := setup(t)
+	backends.NewNeoQTestSuite(n).Run(t)
 }

backends/redis/redis_backend_test.go

@@ -443,7 +443,7 @@ result_loop:
 	}
 }
 
-func TestFutureJobSchedulingOverride(t *testing.T) {
+func TestSuite(t *testing.T) {
 	connString := os.Getenv("TEST_REDIS_URL")
 	if connString == "" {
 		t.Skip("Skipping: TEST_REDIS_URL not set")
@@ -461,6 +461,6 @@ func TestFutureJobSchedulingOverride(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer nq.Shutdown(ctx)
-	backends.TestOverrideFingerprint(t, ctx, nq)
+
+	backends.NewNeoQTestSuite(nq).Run(t)
 }

backends/suite.go

@@ -0,0 +1,30 @@
+package backends
+
+import (
+	"context"
+	"testing"
+
+	"github.com/acaloiaro/neoq"
+	"github.com/stretchr/testify/suite"
+)
+
+type NeoQTestSuite struct {
+	suite.Suite
+	NeoQ neoq.Neoq
+}
+
+// NewNeoQTestSuite constructs a new NeoQ test suite that can be used to test
+// any impementation of the queue
+func NewNeoQTestSuite(q neoq.Neoq) *NeoQTestSuite {
+	n := new(NeoQTestSuite)
+	n.NeoQ = q
+	return n
+}
+
+func (s *NeoQTestSuite) Run(t *testing.T) {
+	suite.Run(t, s)
+}
+
+func (s *NeoQTestSuite) TearDownSuite() {
+	s.NeoQ.Shutdown(context.Background())
+}