Add migration notice to README

Bumps [golang.org/x/crypto](https://github.com/golang/crypto) from 0.31.0 to 0.35.0.
- [Commits](https://github.com/golang/crypto/compare/v0.31.0...v0.35.0)

---
updated-dependencies:
- dependency-name: golang.org/x/crypto
  dependency-version: 0.35.0
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>

.github/workflows/golangci-lint.yml

@@ -12,17 +12,17 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/setup-go@v3
-        with:
-          go-version: '1.21'
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: go mod
         run: make mod
+      - uses: actions/setup-go@v5
+        with:
+          go-version: stable
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3
+        uses: golangci/golangci-lint-action@v8
         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: v1.51
+          version: latest
 
           # Optional: working directory, useful for monorepos
           # working-directory: somedir

.github/workflows/test.yml

@@ -34,7 +34,7 @@ jobs:
           --health-timeout 5s
           --health-retries 5
 
-    container: golang:1.21
+    container: golang:1.24.0
 
     steps:
       - uses: actions/checkout@v3
@@ -44,7 +44,7 @@ jobs:
           TEST_DATABASE_URL: postgres://postgres:postgres@postgres:5432/postgres?sslmode=disable&pool_max_conns=8
           TEST_REDIS_URL: redis:6379
       - name: upload results
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
           name: Upload test coverage
           path: tmp/coverage

.golangci.yaml

@@ -1,103 +1,12 @@
----
+version: "2"
 run:
   concurrency: 12
-  timeout: 1m
+  build-tags:
+    - testing
+  modules-download-mode: vendor
   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: []
-
-  # 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
-  # automatic updating of go.mod described above. Instead, it fails when any changes
-  # to go.mod are needed. This setting is most useful to check that go.mod does
-  # not need updates, such as in a continuous integration and testing system.
-  # If invoked with -mod=vendor, the go command assumes that the vendor
-  # directory holds the correct copies of dependencies and ignores
-  # the dependency descriptions in go.mod.
-  modules-download-mode: vendor
-
-# 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:
-
-linters-settings:
-  dupl:
-    threshold: 100
-  funlen:
-    lines: 100
-    statements: 50
-  goconst:
-    min-len: 2
-    min-occurrences: 2
-  gocritic:
-    enabled-tags:
-      - diagnostic
-      - experimental
-      - opinionated
-      - performance
-      - style
-    disabled-checks:
-      - ifElseChain
-      - octalLiteral
-      - unnamedResult
-      - whyNoLint
-      - wrapperFunc
-  gocyclo:
-    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:
-          - (github.com/golangci/golangci-lint/pkg/logutils.Log).Infof
-          - (github.com/golangci/golangci-lint/pkg/logutils.Log).Warnf
-          - (github.com/golangci/golangci-lint/pkg/logutils.Log).Errorf
-          - (github.com/golangci/golangci-lint/pkg/logutils.Log).Fatalf
-  lll:
-    line-length: 140
-  misspell:
-    locale: US
-
 linters:
-  # please, do not use `enable-all`: it's deprecated and will be removed soon.
-  # inverted configuration with `enable-all` and `disable` is not scalable during updates of golangci-lint
-  #disable-all: true
   enable:
     - asasalint
     - asciicheck
@@ -111,41 +20,30 @@ linters:
     - dupl
     - dupword
     - durationcheck
-    - errcheck
+    - err113
     - errchkjson
     - errname
     - errorlint
-    - execinquery
     - exhaustive
-    - exportloopref
     - funlen
-    - gci
     - gocognit
     - goconst
     - gocritic
     - gocyclo
-    # - godox
-    - goerr113
-    - gofmt
     - goheader
-    - goimports
-    - gomnd
     - gomoddirectives
     - gomodguard
     - goprintffuncname
     - gosec
-    - gosimple
-    - govet
     - grouper
     - importas
-    - ineffassign
     - interfacebloat
-    #- ireturn
     - lll
     - loggercheck
     - maintidx
     - makezero
     - misspell
+    - mnd
     - nestif
     - nilerr
     - nilnil
@@ -159,32 +57,80 @@ linters:
     - rowserrcheck
     - sqlclosecheck
     - staticcheck
-    - stylecheck
-    - tenv
     - testableexamples
-    #- testpackage
     - thelper
     - unconvert
     - unparam
-    - unused
     - usestdlibvars
     - wastedassign
     - whitespace
     - wrapcheck
-
-  # don't enable:
-
-issues:
-  exclude:
-    - 'shadow: declaration of "err" shadows declaration at'
-  # Excluding configuration per-path, per-linter, per-text and per-source
-  exclude-rules:
-    - path: _test\.go
-      linters:
-        - 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
+  settings:
+    dupl:
+      threshold: 100
+    funlen:
+      lines: 100
+      statements: 50
+    goconst:
+      min-len: 2
+      min-occurrences: 2
+    gocritic:
+      disabled-checks:
+        - ifElseChain
+        - octalLiteral
+        - unnamedResult
+        - whyNoLint
+        - wrapperFunc
+      enabled-tags:
+        - diagnostic
+        - experimental
+        - opinionated
+        - performance
+        - style
+    gocyclo:
+      min-complexity: 15
+    govet:
+      settings:
+        printf:
+          funcs:
+            - (github.com/golangci/golangci-lint/pkg/logutils.Log).Infof
+            - (github.com/golangci/golangci-lint/pkg/logutils.Log).Warnf
+            - (github.com/golangci/golangci-lint/pkg/logutils.Log).Errorf
+            - (github.com/golangci/golangci-lint/pkg/logutils.Log).Fatalf
+    lll:
+      line-length: 140
+    misspell:
+      locale: US
+  exclusions:
+    generated: lax
+    presets:
+      - comments
+      - common-false-positives
+      - legacy
+      - std-error-handling
+    rules:
+      - linters:
+          - cyclop
+          - funlen
+        path: _test\.go
+      - path: (.+)\.go$
+        text: 'shadow: declaration of "err" shadows declaration at'
+    paths:
+      - third_party$
+      - builtin$
+      - examples$
+formatters:
+  enable:
+    - gci
+    - gofmt
+    - goimports
+  settings:
+    goimports:
+      local-prefixes:
+        - github.com/golangci/golangci-lint
+  exclusions:
+    generated: lax
+    paths:
+      - third_party$
+      - builtin$
+      - examples$

README.md

@@ -1,3 +1,5 @@
+**NOTICE**: This repository has been moved to [https://code.adriano.fyi/me/neoq](https://code.adriano.fyi/me/neoq). Microsoft has shown little interest in stewarding what was once the best and largest open source community. This small act of migration is my way showing that we don't all support Microsoft's disinterest.
+
 # Neoq
 
 Queue-agnostic background job library for Go, with a pleasant API and powerful features.
@@ -164,4 +166,4 @@ Run `make test` to run the test suite.
 
 # Status
 
-This project is currently pre-1.0. Future releases may change the API.
+This project is currently pre-1.0. Future releases may change the API.

backends/memory/memory_backend.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"os"
 	"sync"
 	"time"
@@ -17,7 +18,6 @@ import (
 	"github.com/iancoleman/strcase"
 	"github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
 	"github.com/robfig/cron"
-	"golang.org/x/exp/slog"
 )
 
 const (

backends/memory/memory_backend_test.go

@@ -3,6 +3,7 @@ package memory_test
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"os"
 	"strconv"
 	"strings"
@@ -18,7 +19,6 @@ import (
 	"github.com/acaloiaro/neoq/testutils"
 	"github.com/pkg/errors"
 	"github.com/robfig/cron"
-	"golang.org/x/exp/slog"
 )
 
 const (
@@ -75,7 +75,7 @@ func TestBasicJobProcessing(t *testing.T) {
 				},
 			})
 			if e != nil || jid == jobs.DuplicateJobID {
-				slog.Error("job was not enqueued. either it was duplicate or this error caused it:", e)
+				slog.Error("job was not enqueued. either it was duplicate or this error caused it:", "error", e)
 			}
 		}
 	}()
@@ -149,7 +149,7 @@ func TestBackendConfiguration(t *testing.T) {
 				},
 			})
 			if e != nil || jid == jobs.DuplicateJobID {
-				slog.Error("job was not enqueued. either it was duplicate or this error caused it:", e)
+				slog.Error("job was not enqueued. either it was duplicate or this error caused it:", "error", e)
 			}
 		}
 

backends/postgres/postgres_backend.go

@@ -5,8 +5,11 @@ import (
 	"embed"
 	"errors"
 	"fmt"
+	"log/slog"
+	"maps"
 	"net/url"
 	"os"
+	"slices"
 	"sync"
 	"time"
 
@@ -26,8 +29,6 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
 	"github.com/robfig/cron"
-	"golang.org/x/exp/slices"
-	"golang.org/x/exp/slog"
 )
 
 //go:embed migrations/*.sql
@@ -46,6 +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 = ANY($1)
 					AND run_after <= NOW()
 					ORDER BY created_at ASC
 					FOR UPDATE SKIP LOCKED
@@ -220,16 +222,13 @@ func (p *PgBackend) listenerManager(ctx context.Context) {
 			p.listenerConnMu.Lock()
 			p.listenerConn = lc
 			p.mu.Lock()
-			handlers := p.handlers
-			p.mu.Unlock()
-
-			for queue := range handlers {
+			for queue := range p.handlers {
 				_, err = p.listenerConn.Exec(ctx, fmt.Sprintf(`LISTEN %q`, queue))
 				if err != nil {
 					p.logger.Error("unable to listen on queue", slog.Any("error", err), slog.String("queue", queue))
 				}
 			}
-
+			p.mu.Unlock()
 			p.listenerConnMu.Unlock()
 
 			p.logger.Debug("worker database connection established")
@@ -835,6 +834,9 @@ func (p *PgBackend) announceJob(ctx context.Context, queue, jobID string) {
 
 // processPendingJobs starts a goroutine that periodically fetches pendings jobs and announces them to workers.
 //
+// The interval for this task does not need to be particularly short unless an application suffers frequent database
+// disconnects, as most pending jobs are picked up by LISTEN after being announced with NOTIFY.
+//
 // Past due jobs are fetched on the interval [neoq.DefaultPendingJobFetchInterval]
 // nolint: cyclop
 func (p *PgBackend) processPendingJobs(ctx context.Context) {
@@ -842,7 +844,12 @@ func (p *PgBackend) processPendingJobs(ctx context.Context) {
 		var err error
 		var conn *pgxpool.Conn
 		var pendingJobs []*jobs.Job
-		ticker := time.NewTicker(neoq.DefaultPendingJobFetchInterval)
+		var ticker *time.Ticker
+		if p.config.PendingJobCheckInterval > 0 {
+			ticker = time.NewTicker(p.config.PendingJobCheckInterval)
+		} else {
+			ticker = time.NewTicker(neoq.DefaultPendingJobFetchInterval)
+		}
 
 		// check for pending jobs on an interval until the context is canceled
 		for {
@@ -1044,7 +1051,11 @@ func (p *PgBackend) getJob(ctx context.Context, tx pgx.Tx, jobID string) (job *j
 }
 
 func (p *PgBackend) getPendingJobs(ctx context.Context, conn *pgxpool.Conn) (pendingJobs []*jobs.Job, err error) {
-	rows, err := conn.Query(ctx, PendingJobsQuery)
+	p.mu.Lock()
+	// 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, activeQueues)
 	if err != nil {
 		return
 	}

backends/postgres/postgres_backend_test.go

@@ -1138,42 +1138,58 @@ 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)
 
 	ctx := context.Background()
 
-	// INSERTing jobs into 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 {
-		t.Error(fmt.Errorf("unable to add job to queue: %w", err))
-		return
-	}
-
-	nq, err := neoq.New(ctx, neoq.WithBackend(postgres.Backend), postgres.WithConnectionString(connString))
+	nq, err := neoq.New(ctx,
+		neoq.WithBackend(postgres.Backend),
+		postgres.WithConnectionString(connString),
+		neoq.WithPendingJobCheckInterval(50*time.Millisecond),
+	)
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer nq.Shutdown(ctx)
 
+	// Adding jobs into 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
+	pendingJobID, err = nq.Enqueue(ctx, &jobs.Job{
+		Queue:   queue,
+		Payload: payload,
+	})
+	if err != nil {
+		t.Error(fmt.Errorf("unable to add job to queue: %w", err))
+		return
+	}
+
 	h := handler.New(queue, func(_ context.Context) (err error) {
 		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

backends/redis/redis_backend.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"log/slog"
 	"os"
 	"runtime"
 	"strings"
@@ -20,7 +21,6 @@ import (
 	"github.com/hibiken/asynq"
 	"github.com/iancoleman/strcase"
 	"github.com/jsuar/go-cron-descriptor/pkg/crondescriptor"
-	"golang.org/x/exp/slog"
 )
 
 // All jobs are placed on the same 'default' queue (until a compelling case is made for using different asynq queues

flake.lock

@@ -207,11 +207,11 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1705697961,
-        "narHash": "sha256-XepT3WS516evSFYkme3GrcI3+7uwXHqtHbip+t24J7E=",
+        "lastModified": 1750994206,
+        "narHash": "sha256-3u6rEbIX9CN/5A5/mc3u0wIO1geZ0EhjvPBXmRDHqWM=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "e5d1c87f5813afde2dda384ac807c57a105721cc",
+        "rev": "80d50fc87924c2a0d346372d242c27973cf8cdbf",
         "type": "github"
       },
       "original": {

flake.nix

@@ -32,7 +32,7 @@
             packages = with pkgs; [
               automake
               gcc
-              go_1_21
+              go_1_23
               gomod2nix.legacyPackages.${system}.gomod2nix
               gotools
               golangci-lint
@@ -57,7 +57,7 @@
 
             services = {
               postgres = {
-                package = pkgs.postgresql;
+                package = pkgs.postgresql_15;
                 enable = true;
                 listen_addresses = "127.0.0.1";
                 port = postgresPort;

go.mod

@@ -1,6 +1,8 @@
 module github.com/acaloiaro/neoq
 
-go 1.21
+go 1.23.0
+
+toolchain go1.24.1
 
 require (
 	github.com/golang-migrate/migrate/v4 v4.16.2
@@ -12,7 +14,6 @@ require (
 	github.com/jsuar/go-cron-descriptor v0.1.0
 	github.com/pkg/errors v0.9.1
 	github.com/robfig/cron v1.2.0
-	golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1
 )
 
 require (
@@ -32,10 +33,10 @@ require (
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
-	golang.org/x/crypto v0.31.0 // indirect
-	golang.org/x/sync v0.10.0 // indirect
-	golang.org/x/sys v0.28.0 // indirect
-	golang.org/x/text v0.21.0 // indirect
+	golang.org/x/crypto v0.35.0 // indirect
+	golang.org/x/sync v0.11.0 // indirect
+	golang.org/x/sys v0.30.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
 	golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
 )

go.sum

@@ -154,11 +154,9 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
-golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
+golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs=
+golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1 h1:MGwJjxBy0HJshjDNfLsYO8xppfqWlA5ZT9OhtUUhTNw=
-golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -187,8 +185,8 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
-golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -202,13 +200,13 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
-golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
-golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

gomod2nix.toml

@@ -79,9 +79,6 @@ schema = 3
   [mod."golang.org/x/crypto"]
     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.10.0"
     hash = "sha256-HWruKClrdoBKVdxKCyoazxeQV4dIYLdkHekQvx275/o="

handler/handler.go

@@ -4,12 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"runtime"
 	"runtime/debug"
 	"strings"
 	"time"
-
-	"golang.org/x/exp/slog"
 )
 
 const (

logging/logging.go

@@ -1,7 +1,7 @@
 package logging
 
 import (
-	"golang.org/x/exp/slog"
+	"log/slog"
 )
 
 // Logger interface is the interface that neoq's logger must implement

neoq.go

@@ -18,7 +18,7 @@ const (
 	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
+	// E.g. right now is 16:00:00 and a job's RunAfter is 16:00:30 of the same date. This job will get a dedicated goroutine
 	// to wait until the job's RunAfter, scheduling the job to be run exactly at RunAfter
 	DefaultFutureJobWindow  = 30 * time.Second
 	DefaultJobCheckInterval = 1 * time.Second
@@ -32,18 +32,19 @@ var ErrBackendNotSpecified = errors.New("a backend must be specified")
 // backends. [BackendConcurrency], for example, is only used by the redis backend. Other backends manage concurrency on a
 // per-handler basis.
 type Config struct {
-	BackendInitializer     BackendInitializer
-	BackendAuthPassword    string                   // password with which to authenticate to the backend
-	BackendConcurrency     int                      // total number of backend processes available to process jobs
-	ConnectionString       string                   // a string containing connection details for the backend
-	JobCheckInterval       time.Duration            // the interval of time between checking for new future/past-due jobs
-	FutureJobWindow        time.Duration            // time duration between current time and job.RunAfter that future jobs get scheduled
-	IdleTransactionTimeout int                      // number of milliseconds PgBackend transaction may idle before the connection is killed
-	ShutdownTimeout        time.Duration            // duration to wait for jobs to finish during shutdown
-	SynchronousCommit      bool                     // Postgres: Enable synchronous commits (increases durability, decreases performance)
-	LogLevel               logging.LogLevel         // the log level of the default logger
-	PGConnectionTimeout    time.Duration            // the amount of time to wait for a connection to become available before timing out
-	RecoveryCallback       handler.RecoveryCallback // the recovery handler applied to all Handlers excuted by the associated Neoq instance
+	BackendInitializer      BackendInitializer
+	BackendAuthPassword     string                   // password with which to authenticate to the backend
+	BackendConcurrency      int                      // total number of backend processes available to process jobs
+	ConnectionString        string                   // a string containing connection details for the backend
+	JobCheckInterval        time.Duration            // the interval of time between checking for new future jobs
+	PendingJobCheckInterval time.Duration            // duration of time between checking for unprocessed jobs due to downtime/db disconnects
+	FutureJobWindow         time.Duration            // time duration between current time and job.RunAfter that future jobs get scheduled
+	IdleTransactionTimeout  int                      // number of milliseconds PgBackend transaction may idle before the connection is killed
+	ShutdownTimeout         time.Duration            // duration to wait for jobs to finish during shutdown
+	SynchronousCommit       bool                     // Postgres: Enable synchronous commits (increases durability, decreases performance)
+	LogLevel                logging.LogLevel         // the log level of the default logger
+	PGConnectionTimeout     time.Duration            // the amount of time to wait for a connection to become available before timing out
+	RecoveryCallback        handler.RecoveryCallback // the recovery handler applied to all Handlers excuted by the associated Neoq instance
 }
 
 // ConfigOption is a function that sets optional backend configuration
@@ -139,6 +140,17 @@ func WithJobCheckInterval(interval time.Duration) ConfigOption {
 	}
 }
 
+// WithPendingJobCheckInterval configures the duration of time between checking for unprocessed jobs on watched queues.
+//
+// Pending jobs are jobs that were queued while neoq wasn't listening or while a disconnect occurred. Most jobs are
+// processed via Postgres' LISTEN/NOTIFY facilities. This check interval is a last resort and does not need to be short.
+// Be careful not to hammer your database by setting this too low.
+func WithPendingJobCheckInterval(interval time.Duration) ConfigOption {
+	return func(c *Config) {
+		c.PendingJobCheckInterval = interval
+	}
+}
+
 // WithLogLevel configures the log level for neoq's default logger. By default, log level is "INFO".
 // if SetLogger is used, WithLogLevel has no effect on the set logger
 func WithLogLevel(level logging.LogLevel) ConfigOption {