Upgrade to golangci-lint v2.2.1 + disable some annoying lints + fix others

.github/workflows/ci.yaml

@@ -243,7 +243,7 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     env:
-      GOLANGCI_LINT_VERSION: v2.1.6
+      GOLANGCI_LINT_VERSION: v2.2.1
     permissions:
       contents: read
       # allow read access to pull request. Use with `only-new-issues` option.

.golangci.yaml

@@ -23,10 +23,12 @@ linters:
     - maintidx # ANOTHER ANOTHER "cyclomatic complexity" lint (see also "cyclop" and "gocyclo")
     - mnd # detects "magic numbers", which it defines as any number; annoying
     - nestif # yells when if blocks are nested; what planet do these people come from?
+    - noinlineerr # disallows `if err := ...`; because why miss an opportunity to leak variables out of scope?
     - ireturn # bans returning interfaces; questionable as is, but also buggy as hell; very, very annoying
     - lll # restricts maximum line length; annoying
     - nlreturn # requires a blank line before returns; annoying
     - wsl # a bunch of style/whitespace stuff; annoying
+    - wsl_v5 # a second version of the first annoying wsl; how nice
 
   settings:
     depguard:

client.go

@@ -471,10 +471,10 @@ func (c *Config) validate() error {
 	// this convention.
 	maxSchemaLength := 63 - 1 - len(string(notifier.NotificationTopicLongest)) // -1 for the dot in `<schema>.<topic>`
 	if len(c.Schema) > maxSchemaLength {
-		return fmt.Errorf("Schema length must be less than or equal to %d characters", maxSchemaLength) //nolint:staticcheck
+		return fmt.Errorf("Schema length must be less than or equal to %d characters", maxSchemaLength)
 	}
 	if c.Schema != "" && !postgresSchemaNameRE.MatchString(c.Schema) {
-		return errors.New("Schema name can only contain letters, numbers, and underscores, and must start with a letter or underscore") //nolint:staticcheck
+		return errors.New("Schema name can only contain letters, numbers, and underscores, and must start with a letter or underscore")
 	}
 
 	for queue, queueConfig := range c.Queues {
@@ -492,7 +492,7 @@ func (c *Config) validate() error {
 			kind := workerInfo.jobArgs.Kind()
 			if !rivercommon.UserSpecifiedIDOrKindRE.MatchString(kind) {
 				if c.SkipJobKindValidation {
-					c.Logger.Warn("job kind should match regex; this will be an error in future versions",
+					c.Logger.Warn("job kind should match regex; this will be an error in future versions", //nolint:noctx
 						slog.String("kind", kind),
 						slog.String("regex", rivercommon.UserSpecifiedIDOrKindRE.String()),
 					)
@@ -763,7 +763,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 				client.services = append(client.services, client.notifier)
 			}
 		} else {
-			config.Logger.Info("Driver does not support listener; entering poll only mode")
+			config.Logger.Info("Driver does not support listener; entering poll only mode") //nolint:noctx
 		}
 
 		client.elector = leadership.NewElector(archetype, driver.GetExecutor(), client.notifier, &leadership.Config{

client_test.go

@@ -1027,6 +1027,7 @@ func Test_Client_Common(t *testing.T) {
 
 		type JobArgs struct {
 			testutil.JobArgsReflectKind[JobArgs]
+
 			Name string `json:"name"`
 		}
 
@@ -1452,6 +1453,7 @@ var (
 
 type workerWithMiddleware[T JobArgs] struct {
 	WorkerDefaults[T]
+
 	workFunc       func(context.Context, *Job[T]) error
 	middlewareFunc func(*rivertype.JobRow) []rivertype.WorkerMiddleware
 }
@@ -1890,6 +1892,7 @@ func (callbackWithCustomTimeoutArgs) Kind() string { return "callbackWithCustomT
 
 type callbackWorkerWithCustomTimeout struct {
 	WorkerDefaults[callbackWithCustomTimeoutArgs]
+
 	fn func(context.Context, *Job[callbackWithCustomTimeoutArgs]) error
 }
 
@@ -5674,6 +5677,7 @@ func Test_Client_Subscribe(t *testing.T) {
 
 	type JobArgs struct {
 		testutil.JobArgsReflectKind[JobArgs]
+
 		Name string `json:"name"`
 	}
 
@@ -5948,6 +5952,7 @@ func Test_Client_SubscribeConfig(t *testing.T) {
 
 		type JobArgs struct {
 			testutil.JobArgsReflectKind[JobArgs]
+
 			Name string `json:"name"`
 		}
 
@@ -6251,7 +6256,7 @@ func Test_Client_InsertNotificationsAreDeduplicatedAndDebounced(t *testing.T) {
 	}
 	notifyCh := make(chan notification, 10)
 	handleNotification := func(topic notifier.NotificationTopic, payload string) {
-		config.Logger.Info("received notification", slog.String("topic", string(topic)), slog.String("payload", payload))
+		config.Logger.InfoContext(ctx, "received notification", slog.String("topic", string(topic)), slog.String("payload", payload))
 		notif := notification{topic: topic}
 		require.NoError(t, json.Unmarshal([]byte(payload), &notif.payload))
 		notifyCh <- notif
@@ -6262,7 +6267,7 @@ func Test_Client_InsertNotificationsAreDeduplicatedAndDebounced(t *testing.T) {
 
 	expectImmediateNotification := func(t *testing.T, queue string) {
 		t.Helper()
-		config.Logger.Info("inserting " + queue + " job")
+		config.Logger.InfoContext(ctx, "inserting "+queue+" job")
 		_, err = client.Insert(ctx, JobArgs{}, &InsertOpts{Queue: queue})
 		require.NoError(t, err)
 		notif := riversharedtest.WaitOrTimeout(t, notifyCh)
@@ -6275,7 +6280,7 @@ func Test_Client_InsertNotificationsAreDeduplicatedAndDebounced(t *testing.T) {
 	tNotif1 := time.Now()
 
 	for range 5 {
-		config.Logger.Info("inserting queue1 job")
+		config.Logger.InfoContext(ctx, "inserting queue1 job")
 		_, err = client.Insert(ctx, JobArgs{}, &InsertOpts{Queue: "queue1"})
 		require.NoError(t, err)
 	}
@@ -7280,6 +7285,7 @@ type testWorkerDeadline struct {
 
 type timeoutTestWorker struct {
 	WorkerDefaults[timeoutTestArgs]
+
 	doneCh chan testWorkerDeadline
 }
 
@@ -7548,6 +7554,7 @@ func TestInsertParamsFromJobArgsAndOptions(t *testing.T) {
 
 		type PartialArgs struct {
 			JobArgsStaticKind
+
 			Included bool `json:"included" river:"unique"`
 			Excluded bool `json:"excluded"`
 		}

example_complete_job_within_tx_test.go

@@ -26,6 +26,7 @@ func (TransactionalArgs) Kind() string { return "transactional_worker" }
 // the transaction such as inserting additional jobs or manipulating other data.
 type TransactionalWorker struct {
 	river.WorkerDefaults[TransactionalArgs]
+
 	dbPool *pgxpool.Pool
 }
 

example_graceful_shutdown_test.go

@@ -28,6 +28,7 @@ func (WaitsForCancelOnlyArgs) Kind() string { return "waits_for_cancel_only" }
 // context is cancelled.
 type WaitsForCancelOnlyWorker struct {
 	river.WorkerDefaults[WaitsForCancelOnlyArgs]
+
 	jobStarted chan struct{}
 }
 

example_job_cancel_from_client_test.go

@@ -22,6 +22,7 @@ func (args SleepingArgs) Kind() string { return "SleepingWorker" }
 
 type SleepingWorker struct {
 	river.WorkerDefaults[CancellingArgs]
+
 	jobChan chan int64
 }
 

example_queue_pause_test.go

@@ -22,6 +22,7 @@ func (args ReportingArgs) Kind() string { return "Reporting" }
 
 type ReportingWorker struct {
 	river.WorkerDefaults[ReportingArgs]
+
 	jobWorkedCh chan<- string
 }
 

internal/dbunique/db_unique_test.go

@@ -43,6 +43,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type EmailJobArgs struct {
 					JobArgsStaticKind
+
 					Recipient   string `json:"recipient"    river:"unique"`
 					Subject     string `json:"subject"      river:"unique"`
 					Body        string `json:"body"`
@@ -66,6 +67,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type SMSJobArgs struct {
 					JobArgsStaticKind
+
 					PhoneNumber string `json:"phone_number"      river:"unique"`
 					Message     string `json:"message,omitempty" river:"unique"`
 					TemplateID  int    `json:"template_id"`
@@ -85,6 +87,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type EmailJobArgs struct {
 					JobArgsStaticKind
+
 					Recipient  string `river:"unique"`
 					Subject    string `river:"unique"`
 					TemplateID int
@@ -104,6 +107,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type EmailJobArgs struct {
 					JobArgsStaticKind
+
 					Recipient string `json:"recipient" river:"unique"`
 					Subject   string `json:"subject"   river:"unique"`
 					Body      string `json:"body"`
@@ -123,6 +127,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type GenericJobArgs struct {
 					JobArgsStaticKind
+
 					Description string `json:"description"`
 					Count       int    `json:"count"`
 					foo         string // won't be marshaled in JSON
@@ -158,6 +163,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type TaskJobArgs struct {
 					JobArgsStaticKind
+
 					TaskID string
 				}
 				return TaskJobArgs{
@@ -189,6 +195,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type TaskJobArgs struct {
 					JobArgsStaticKind
+
 					TaskID string `json:"task_id"`
 				}
 				return TaskJobArgs{
@@ -204,6 +211,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type TaskJobArgs struct {
 					JobArgsStaticKind
+
 					TaskID string `json:"task_id"`
 				}
 				return TaskJobArgs{
@@ -219,6 +227,7 @@ func TestUniqueKey(t *testing.T) {
 			argsFunc: func() rivertype.JobArgs {
 				type TaskJobArgs struct {
 					JobArgsStaticKind
+
 					TaskID string `json:"task_id"`
 				}
 				return TaskJobArgs{

internal/jobcompleter/job_completer.go

@@ -225,7 +225,7 @@ func (c *AsyncCompleter) Start(ctx context.Context) error {
 		<-ctx.Done()
 
 		if err := c.errGroup.Wait(); err != nil {
-			c.Logger.Error("Error waiting on async completer", "err", err)
+			c.Logger.ErrorContext(ctx, "Error waiting on async completer", "err", err)
 		}
 	}()
 
@@ -316,7 +316,7 @@ func (c *BatchCompleter) Start(ctx context.Context) error {
 				// Try to insert last batch before leaving. Note we use the
 				// original context so operations aren't immediately cancelled.
 				if err := c.handleBatch(ctx); err != nil {
-					c.Logger.Error(c.Name+": Error completing batch", "err", err)
+					c.Logger.ErrorContext(ctx, c.Name+": Error completing batch", "err", err)
 				}
 				return
 
@@ -336,7 +336,7 @@ func (c *BatchCompleter) Start(ctx context.Context) error {
 
 			for {
 				if err := c.handleBatch(ctx); err != nil {
-					c.Logger.Error(c.Name+": Error completing batch", "err", err)
+					c.Logger.ErrorContext(ctx, c.Name+": Error completing batch", "err", err)
 				}
 
 				// New jobs to complete may have come in while working the batch

internal/jobcompleter/job_completer_test.go

@@ -28,6 +28,7 @@ import (
 
 type partialExecutorMock struct {
 	riverdriver.Executor
+
 	JobSetStateIfRunningManyCalled bool
 	JobSetStateIfRunningManyFunc   func(ctx context.Context, params *riverdriver.JobSetStateIfRunningManyParams) ([]*rivertype.JobRow, error)
 	mu                             sync.Mutex
@@ -63,6 +64,7 @@ func (m *partialExecutorMock) setCalled(setCalledFunc func()) {
 
 type partialExecutorTxMock struct {
 	riverdriver.ExecutorTx
+
 	partial *partialExecutorMock
 }
 

internal/jobexecutor/job_executor.go

@@ -124,8 +124,8 @@ type JobExecutor struct {
 	stats *jobstats.JobStatistics // initialized by the executor, and handed off to completer
 }
 
-func (e *JobExecutor) Cancel() {
-	e.Logger.Warn(e.Name+": job cancelled remotely", slog.Int64("job_id", e.JobRow.ID))
+func (e *JobExecutor) Cancel(ctx context.Context) {
+	e.Logger.WarnContext(ctx, e.Name+": job cancelled remotely", slog.Int64("job_id", e.JobRow.ID))
 	e.CancelFunc(rivertype.ErrJobCancelledRemotely)
 }
 

internal/jobexecutor/job_executor_test.go

@@ -761,7 +761,7 @@ func TestJobExecutor_Execute(t *testing.T) {
 
 		go func() {
 			<-jobStarted
-			executor.Cancel()
+			executor.Cancel(ctx)
 			close(haveCancelled)
 		}()
 

internal/leadership/elector.go

@@ -350,7 +350,7 @@ func (e *Elector) keepLeadershipLoop(ctx context.Context) error {
 			}
 
 			sleepDuration := serviceutil.ExponentialBackoff(numErrors, serviceutil.MaxAttemptsBeforeResetDefault)
-			e.Logger.Error(e.Name+": Error attempting reelection", e.errorSlogArgs(err, numErrors, sleepDuration)...)
+			e.Logger.ErrorContext(ctx, e.Name+": Error attempting reelection", e.errorSlogArgs(err, numErrors, sleepDuration)...)
 			serviceutil.CancellableSleep(ctx, sleepDuration)
 			continue
 		}

internal/maintenance/job_cleaner.go

@@ -172,7 +172,7 @@ func (s *JobCleaner) runOnce(ctx context.Context) (*jobCleanerRunOnceResult, err
 				Schema:                      s.Config.Schema,
 			})
 			if err != nil {
-				return 0, fmt.Errorf("error deleting completed jobs: %w", err)
+				return 0, fmt.Errorf("error cleaning jobs: %w", err)
 			}
 
 			return numDeleted, nil

internal/maintenance/queue_maintainer.go

@@ -112,6 +112,7 @@ func serviceName(service startstop.Service) string {
 // should be called on service start to avoid thundering herd problems.
 type queueMaintainerServiceBase struct {
 	baseservice.BaseService
+
 	staggerStartupDisabled bool
 }
 

internal/riverinternaltest/sharedtx/shared_tx.go

@@ -141,6 +141,7 @@ func (d *sharedTxDerivative) unlockParent() {
 // the row finishes scanning.
 type SharedTxRow struct {
 	sharedTxDerivative
+
 	innerRow pgx.Row
 }
 
@@ -153,6 +154,7 @@ func (r *SharedTxRow) Scan(dest ...any) error {
 // the rows are closed.
 type SharedTxRows struct {
 	sharedTxDerivative
+
 	innerRows pgx.Rows
 }
 
@@ -177,6 +179,7 @@ func (r *SharedTxRows) Values() ([]interface{}, error) { return r.innerRows.Valu
 // rolls back.
 type SharedSubTx struct {
 	sharedTxDerivative
+
 	inner pgx.Tx
 }
 

periodic_job_test.go

@@ -54,6 +54,7 @@ func TestPeriodicJobBundle(t *testing.T) {
 
 		type TestJobArgs struct {
 			testutil.JobArgsReflectKind[TestJobArgs]
+
 			JobNum int `json:"job_num"`
 		}
 

plugin_test.go

@@ -60,6 +60,7 @@ var _ driverPlugin[pgx.Tx] = &TestDriverWithPlugin{}
 
 type TestDriverWithPlugin struct {
 	*riverpgxv5.Driver
+
 	initCalled bool
 	pilot      riverpilot.Pilot
 }
@@ -134,6 +135,7 @@ var _ pilotPlugin = &TestPilotWithPlugin{}
 
 type TestPilotWithPlugin struct {
 	riverpilot.StandardPilot
+
 	maintenanceService startstop.Service
 	service            startstop.Service
 }