From ef64a8395f41bad21c0609158f37e69b05b4da65 Mon Sep 17 00:00:00 2001 From: Adriano Caloiaro Date: Mon, 17 Feb 2025 08:42:50 -0700 Subject: [PATCH] fix: queued names quoted incorrectly in pending jobs query --- .golangci.yaml | 33 +--------------------- backends/postgres/postgres_backend.go | 5 ++-- backends/postgres/postgres_backend_test.go | 11 ++++++++ flake.nix | 2 +- 4 files changed, 15 insertions(+), 36 deletions(-) diff --git a/.golangci.yaml b/.golangci.yaml index 074284e..5e86192 100644 --- a/.golangci.yaml +++ b/.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 diff --git a/backends/postgres/postgres_backend.go b/backends/postgres/postgres_backend.go index 050c904..7892593 100644 --- a/backends/postgres/postgres_backend.go +++ b/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 } diff --git a/backends/postgres/postgres_backend_test.go b/backends/postgres/postgres_backend_test.go index 836f1c9..20c8bea 100644 --- a/backends/postgres/postgres_backend_test.go +++ b/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 diff --git a/flake.nix b/flake.nix index 44a9f83..668eaee 100644 --- a/flake.nix +++ b/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;