fix: queued names quoted incorrectly in pending jobs query

Author: acaloiaro

.github/workflows/golangci-lint.yml

@@ -22,7 +22,7 @@ jobs:
         uses: golangci/golangci-lint-action@v6
         with:
           # Optional: version of golangci-lint to use in form of v1.2 or v1.2.3 or `latest` to use the latest version
-          version: latest
+          version: v1.64.4
 
           # Optional: working directory, useful for monorepos
           # working-directory: somedir

.golangci.yaml

@@ -5,23 +5,7 @@ run:
   issues-exit-code: 1
   tests: true
   # list of build tags, all linters use it. Default is empty list.
-  build-tags: testing
-
-  # which dirs to skip: issues from them won't be reported;
-  # can use regexp here: generated.*, regexp is applied on full path;
-  # default value is empty list, but default dirs are skipped independently
-  # from this option's value (see skip-dirs-use-default).
-  skip-dirs: []
-
-  # default is true. Enables skipping of directories:
-  #   vendor$, third_party$, testdata$, examples$, Godeps$, builtin$
-  skip-dirs-use-default: true
-
-  # which files to skip: they will be analyzed, but issues from them
-  # won't be reported. Default value is empty list, but there is
-  # no need to include all autogenerated files, we confidently recognize
-  # autogenerated files. If it's not please let us know.
-  skip-files: []
+  build-tags: ['testing']
 
   # by default isn't set. If set we pass it to "go list -mod={option}". From "go help modules":
   # If invoked with -mod=readonly, the go command is disallowed from the implicit
@@ -35,17 +19,12 @@ run:
 
 # output configuration options
 output:
-  # colored-line-number|line-number|json|tab|checkstyle|code-climate, default is "colored-line-number"
-  format: colored-line-number
-
   # print lines of code with issue, default is true
   print-issued-lines: true
 
   # print linter name in the end of issue text, default is true
   print-linter-name: true
 
-  # make issues output unique by line, default is true
-  uniq-by-line: true
 
 #It's a .golangci.yml config file of this repo: we enable more linters than the default and have more strict settings:
 
@@ -75,13 +54,7 @@ linters-settings:
     min-complexity: 15
   goimports:
     local-prefixes: github.com/golangci/golangci-lint
-  gomnd:
-    settings:
-      mnd:
-        # don't include the "operation" and "assign"
-        checks: argument,case,condition,return
   govet:
-    check-shadowing: true
     settings:
       printf:
         funcs:
@@ -181,7 +154,3 @@ issues:
         - funlen
         - cyclop
 
-# golangci.com configuration
-# https://github.com/golangci/golangci/wiki/Configuration
-service:
-  golangci-lint-version: 1.51.x # use the fixed version to not introduce new linters unexpectedly

backends/postgres/postgres_backend.go

@@ -10,7 +10,6 @@ import (
 	"net/url"
 	"os"
 	"slices"
-	"strings"
 	"sync"
 	"time"
 
@@ -48,7 +47,7 @@ const (
 	PendingJobsQuery = `SELECT id,fingerprint,queue,status,deadline,payload,retries,max_retries,run_after,ran_at,created_at,error
 					FROM neoq_jobs
 					WHERE status NOT IN ('processed')
-					AND queue IN ($1)
+					AND QUEUE = ANY($1)
 					AND run_after <= NOW()
 					ORDER BY created_at ASC
 					FOR UPDATE SKIP LOCKED
@@ -1056,7 +1055,7 @@ func (p *PgBackend) getPendingJobs(ctx context.Context, conn *pgxpool.Conn) (pen
 	// convert watched queue map to string: "'queue1', 'queue2', ..." for use in Postgres IN statement
 	activeQueues := slices.Collect(maps.Keys(p.handlers))
 	p.mu.Unlock()
-	rows, err := conn.Query(ctx, PendingJobsQuery, strings.Join(activeQueues, ","))
+	rows, err := conn.Query(ctx, PendingJobsQuery, activeQueues)
 	if err != nil {
 		return
 	}

backends/postgres/postgres_backend_test.go

@@ -1138,6 +1138,7 @@ func TestHandlerRecoveryCallback(t *testing.T) {
 func TestProcessPendingJobs(t *testing.T) {
 	connString, conn := prepareAndCleanupDB(t)
 	const queue = "testing"
+	const queue2 = "testing2"
 	timeoutTimer := time.After(5 * time.Second)
 	done := make(chan bool)
 	defer close(done)
@@ -1173,12 +1174,22 @@ func TestProcessPendingJobs(t *testing.T) {
 		return
 	})
 
+	h2 := handler.New(queue2, 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)
 	}
 
+	// Start listening on a second queued to make sure that pending jobs can be pulled from multiple queues
+	err = nq.Start(ctx, h2)
+	if err != nil {
+		t.Error(err)
+	}
+
 	go func() {
 		var err error
 		var status string

flake.nix

@@ -57,7 +57,7 @@
 
             services = {
               postgres = {
-                package = pkgs.postgresql;
+                package = pkgs.postgresql_15;
                 enable = true;
                 listen_addresses = "127.0.0.1";
                 port = postgresPort;