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
@@ -43,13 +44,14 @@ const (
 					AND status NOT IN ('processed')
 					FOR UPDATE SKIP LOCKED
 					LIMIT 1`
-	PendingJobIDQuery = `SELECT id
+	PendingJobsQuery = `SELECT id,fingerprint,queue,status,deadline,payload,retries,max_retries,run_after,ran_at,created_at,error
 					FROM neoq_jobs
-					WHERE queue = $1
-					AND status NOT IN ('processed')
+					WHERE status NOT IN ('processed')
+					AND QUEUE = ANY($1)
 					AND run_after <= NOW()
+					ORDER BY created_at ASC
 					FOR UPDATE SKIP LOCKED
-					LIMIT 1`
+					LIMIT 100`
 	FutureJobQuery = `SELECT id,fingerprint,queue,status,deadline,payload,retries,max_retries,run_after,ran_at,created_at,error
 					FROM neoq_jobs
 					WHERE queue = $1
@@ -190,6 +192,9 @@ func Backend(ctx context.Context, opts ...neoq.ConfigOption) (pb neoq.Neoq, err
 	// monitor handlers for changes and LISTEN when new queues are added
 	go p.listenerManager(ctx)
 
+	// monitor queues for pending jobs, so neoq is resilient to LISTEN disconnects and reconnections
+	p.processPendingJobs(ctx)
+
 	p.listenConnDown <- true
 
 	p.cron.Start()
@@ -199,7 +204,8 @@ func Backend(ctx context.Context, opts ...neoq.ConfigOption) (pb neoq.Neoq, err
 	return pb, nil
 }
 
-// listenerManager manages the LISTENer connection and adding queue to it
+// listenerManager manages the LISTENer connection and add queues to it
+// nolint: cyclop
 func (p *PgBackend) listenerManager(ctx context.Context) {
 	var err error
 	for {
@@ -216,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")
@@ -539,7 +542,17 @@ func (p *PgBackend) StartCron(ctx context.Context, cronSpec string, h handler.Ha
 
 // SetLogger sets this backend's logger
 func (p *PgBackend) SetLogger(logger logging.Logger) {
+	p.mu.Lock()
 	p.logger = logger
+	p.mu.Unlock()
+}
+
+// Logger gets this backend's logger
+func (p *PgBackend) Logger() (l logging.Logger) {
+	p.mu.Lock()
+	l = p.logger
+	p.mu.Unlock()
+	return
 }
 
 // Shutdown shuts this backend down
@@ -674,8 +687,6 @@ func (p *PgBackend) start(ctx context.Context, h handler.Handler) (err error) {
 		return fmt.Errorf("%w: %s", handler.ErrNoHandlerForQueue, h.Queue)
 	}
 
-	pendingJobsChan := p.pendingJobs(ctx, h.Queue) // process overdue jobs *at startup*
-
 	// wait for the listener to connect and be ready to listen
 	for q := range p.readyQueues {
 		if q == h.Queue {
@@ -688,7 +699,7 @@ func (p *PgBackend) start(ctx context.Context, h handler.Handler) (err error) {
 		p.readyQueues <- q
 	}
 
-	// process all future jobs and retries
+	// process all future jobs
 	go func() { p.scheduleFutureJobs(ctx, h.Queue) }()
 
 	for i := 0; i < h.Concurrency; i++ {
@@ -700,8 +711,6 @@ func (p *PgBackend) start(ctx context.Context, h handler.Handler) (err error) {
 				select {
 				case n = <-listenJobChan:
 					err = p.handleJob(ctx, n.Payload)
-				case n = <-pendingJobsChan:
-					err = p.handleJob(ctx, n.Payload)
 				case <-ctx.Done():
 					return
 				case <-errCh:
@@ -754,7 +763,7 @@ func (p *PgBackend) initFutureJobs(ctx context.Context, queue string) (err error
 	return
 }
 
-// scheduleFutureJobs announces future jobs using NOTIFY on an interval
+// scheduleFutureJobs monitors the future job list for upcoming jobs and announces them to be processed by available workers
 func (p *PgBackend) scheduleFutureJobs(ctx context.Context, queue string) {
 	err := p.initFutureJobs(ctx, queue)
 	if err != nil {
@@ -772,8 +781,8 @@ func (p *PgBackend) scheduleFutureJobs(ctx context.Context, queue string) {
 			if timeUntillRunAfter <= p.config.FutureJobWindow {
 				delete(p.futureJobs, jobID)
 				go func(jid string, j *jobs.Job) {
-					scheduleCh := time.After(timeUntillRunAfter)
-					<-scheduleCh
+					jobDue := time.After(timeUntillRunAfter)
+					<-jobDue
 					p.announceJob(ctx, j.Queue, jid)
 				}(jobID, job)
 			}
@@ -823,42 +832,56 @@ func (p *PgBackend) announceJob(ctx context.Context, queue, jobID string) {
 	}
 }
 
-func (p *PgBackend) pendingJobs(ctx context.Context, queue string) (jobsCh chan *pgconn.Notification) {
-	jobsCh = make(chan *pgconn.Notification)
-
-	conn, err := p.acquire(ctx)
-	if err != nil {
-		p.logger.Error(
-			"failed to acquire database connection to listen for pending queue items",
-			slog.String("queue", queue),
-			slog.Any("error", err),
-		)
-		return
-	}
-
+// 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) {
 	go func(ctx context.Context) {
-		defer conn.Release()
+		var err error
+		var conn *pgxpool.Conn
+		var pendingJobs []*jobs.Job
+		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 {
-			jobID, err := p.getPendingJobID(ctx, conn, queue)
+			conn, err = p.acquire(ctx)
 			if err != nil {
-				if errors.Is(err, pgx.ErrNoRows) || errors.Is(err, context.Canceled) {
-					break
-				}
+				p.Logger().Error("[pending_jobs] unable to get database connection", slog.Any("error", err))
+				<-ticker.C
+				continue
+			}
 
-				p.logger.Error(
-					"failed to fetch pending job",
-					slog.String("queue", queue),
+			pendingJobs, err = p.getPendingJobs(ctx, conn)
+			conn.Release()
+			if errors.Is(err, context.Canceled) {
+				return
+			}
+			if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+				p.Logger().Error(
+					"failed to fetch pending jobs",
 					slog.Any("error", err),
-					slog.String("job_id", jobID),
 				)
-			} else {
-				jobsCh <- &pgconn.Notification{Channel: queue, Payload: jobID}
+			}
+
+			for _, job := range pendingJobs {
+				p.announceJob(ctx, job.Queue, fmt.Sprint(job.ID))
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
 			}
 		}
 	}(ctx)
-
-	return jobsCh
 }
 
 // handleJob is the workhorse of Neoq
@@ -944,9 +967,6 @@ func (p *PgBackend) handleJob(ctx context.Context, jobID string) (err error) {
 }
 
 // listen uses Postgres LISTEN to listen for jobs on a queue
-// TODO: There is currently no handling of listener disconnects in PgBackend.
-// This will lead to jobs not getting processed until the worker is restarted.
-// Implement disconnect handling.
 func (p *PgBackend) listen(ctx context.Context) (c chan *pgconn.Notification, errCh chan error) {
 	c = make(chan *pgconn.Notification)
 	errCh = make(chan error)
@@ -1030,8 +1050,21 @@ func (p *PgBackend) getJob(ctx context.Context, tx pgx.Tx, jobID string) (job *j
 	return
 }
 
-func (p *PgBackend) getPendingJobID(ctx context.Context, conn *pgxpool.Conn, queue string) (jobID string, err error) {
-	err = conn.QueryRow(ctx, PendingJobIDQuery, queue).Scan(&jobID)
+func (p *PgBackend) getPendingJobs(ctx context.Context, conn *pgxpool.Conn) (pendingJobs []*jobs.Job, err error) {
+	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
+	}
+
+	pendingJobs, err = pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[jobs.Job])
+	if err != nil {
+		return
+	}
+
 	return
 }
 
@@ -1042,7 +1075,7 @@ func (p *PgBackend) acquire(ctx context.Context) (conn *pgxpool.Conn, err error)
 	ctx, cancelFunc := context.WithDeadline(ctx, time.Now().Add(p.config.PGConnectionTimeout))
 	defer cancelFunc()
 
-	p.logger.Debug("acquiring connection with timeout", slog.Any("timeout", p.config.PGConnectionTimeout))
+	p.Logger().Debug("acquiring connection with timeout", slog.Any("timeout", p.config.PGConnectionTimeout))
 
 	connCh := make(chan *pgxpool.Conn)
 	errCh := make(chan error)
@@ -1062,7 +1095,7 @@ func (p *PgBackend) acquire(ctx context.Context) (conn *pgxpool.Conn, err error)
 	case err := <-errCh:
 		return nil, err
 	case <-ctx.Done():
-		p.logger.Error("exceeded timeout acquiring a connection from the pool", slog.Any("timeout", p.config.PGConnectionTimeout))
+		p.Logger().Error("exceeded timeout acquiring a connection from the pool", slog.Any("timeout", p.config.PGConnectionTimeout))
 		cancelFunc()
 		err = ErrExceededConnectionPoolTimeout
 		return

backends/postgres/postgres_backend_test.go

@@ -1132,3 +1132,91 @@ func TestHandlerRecoveryCallback(t *testing.T) {
 		t.Error(err)
 	}
 }
+
+// TestProcessPendingJobs tests that unanounced jobs with a run_after before the current timestamp get run periodically
+// This ensures that when LISTENER connections fails, that jobs that would have been announced still get processed eventually
+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()
+
+	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
+		// ensure job has failed/has the correct status
+		for {
+			err = conn.
+				QueryRow(context.Background(), "SELECT status FROM neoq_jobs WHERE id = $1", pendingJobID).
+				Scan(&status)
+			if err != nil {
+				break
+			}
+
+			if status != internal.JobStatusNew {
+				done <- true
+				break
+			}
+
+			time.Sleep(50 * time.Millisecond)
+		}
+	}()
+
+	select {
+	case <-timeoutTimer:
+		err = jobs.ErrJobTimeout
+	case <-done:
+	}
+	if err != nil {
+		t.Errorf("job should have resulted in a status of 'processed', but did not")
+	}
+}

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

@@ -31,7 +31,8 @@
           {
             packages = with pkgs; [
               automake
-              go_1_21
+              gcc
+              go_1_23
               gomod2nix.legacyPackages.${system}.gomod2nix
               gotools
               golangci-lint
@@ -41,7 +42,7 @@
             ];
 
             enterShell = ''
-              export TEST_DATABASE_URL="postgres://postgres:postgres@localhost:${toString postgresPort}/neoq?sslmode=disable&pool_max_conns=100"
+              export TEST_DATABASE_URL="postgres://postgres:postgres@localhost:${toString postgresPort}/neoq?sslmode=disable&pool_max_conns=250"
               export TEST_REDIS_URL=localhost:${toString redisPort}
               export REDIS_PASSWORD=
             '';
@@ -56,7 +57,7 @@
 
             services = {
               postgres = {
-                package = pkgs.postgresql;
+                package = pkgs.postgresql_15;
                 enable = true;
                 listen_addresses = "127.0.0.1";
                 port = postgresPort;
@@ -64,6 +65,10 @@
                   CREATE USER postgres WITH PASSWORD 'postgres' SUPERUSER;
                   CREATE DATABASE neoq;
                 '';
+                settings = {
+                  max_connections = 250;
+                  log_statement = "all";
+                };
               };
 
               redis = {

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.17.0 // indirect
-	golang.org/x/sync v0.2.0 // indirect
-	golang.org/x/sys v0.15.0 // indirect
-	golang.org/x/text v0.14.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.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
-golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
+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=
@@ -166,8 +164,8 @@ golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHl
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.11.0 h1:bUO06HqtnRcc/7l71XBe4WcqTZ+3AH1J59zWDDwLKgU=
-golang.org/x/mod v0.11.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA=
+golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -179,16 +177,16 @@ golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
-golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
-golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=
+golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 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.2.0 h1:PUR+T4wwASmuSTYdKjYHI5TD22Wy5ogLU5qZCOLxBrI=
-golang.org/x/sync v0.2.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+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.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
-golang.org/x/sys v0.15.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.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
-golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+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=
@@ -222,8 +220,8 @@ golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
-golang.org/x/tools v0.9.1 h1:8WMNJAz3zrtPmnYC7ISf5dEn3MT0gY7jBJfw27yrrLo=
-golang.org/x/tools v0.9.1/go.mod h1:owI94Op576fPu3cIGQeHs3joujW/2Oc6MtlxbF5dfNc=
+golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg=
+golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

gomod2nix.toml

@@ -77,20 +77,17 @@ schema = 3
     version = "v1.24.0"
     hash = "sha256-yLzjFbMWnc5b033gcPLGP0KY1xWPJ3sjnUG/RndmC3o="
   [mod."golang.org/x/crypto"]
-    version = "v0.17.0"
-    hash = "sha256-/vzBaeD/Ymyc7cpjBvSfJfuZ57zWa9LOaZM7b33eIx0="
-  [mod."golang.org/x/exp"]
-    version = "v0.0.0-20230713183714-613f0c0eb8a1"
-    hash = "sha256-VLE9CCOYpTdyBWaQ1YxXpGOBS74wpIR3JJ+JVzNyEkQ="
+    version = "v0.31.0"
+    hash = "sha256-ZBjoG7ZOuTEmjaXPP9txAvjAjC46DeaLs0zrNzi8EQw="
   [mod."golang.org/x/sync"]
-    version = "v0.2.0"
-    hash = "sha256-hKk9zsy2aXY7R0qGFZhGOVvk5qD17f6KHEuK4rGpTsg="
+    version = "v0.10.0"
+    hash = "sha256-HWruKClrdoBKVdxKCyoazxeQV4dIYLdkHekQvx275/o="
   [mod."golang.org/x/sys"]
-    version = "v0.15.0"
-    hash = "sha256-n7TlABF6179RzGq3gctPDKDPRtDfnwPdjNCMm8ps2KY="
+    version = "v0.28.0"
+    hash = "sha256-kzSlDo5FKsQU9cLefIt2dueGUfz9XuEW+mGSGlPATGc="
   [mod."golang.org/x/text"]
-    version = "v0.14.0"
-    hash = "sha256-yh3B0tom1RfzQBf1RNmfdNWF1PtiqxV41jW1GVS6JAg="
+    version = "v0.21.0"
+    hash = "sha256-QaMwddBRnoS2mv9Y86eVC2x2wx/GZ7kr2zAJvwDeCPc="
   [mod."golang.org/x/time"]
     version = "v0.0.0-20190308202827-9d24e82272b4"
     hash = "sha256-azbksMSLQf1CK0jF2i+ESjFenMDR88xRW1tov0metrg="

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

@@ -12,9 +12,13 @@ import (
 
 const (
 	DefaultIdleTxTimeout = 30000
+	// The duration of time between checking if any queues have pending jobs
+	// It's necessary to check for pending jobs periodically because if the LISTENer connection fails at any point, there is a period
+	// of time when new jobs get announced by the new job trigger announcement, but no listeners are LISTENing for work.
+	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
@@ -28,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/retry 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
@@ -135,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 {