From baf299f1b0108ececf547012d57f2bf309b58462 Mon Sep 17 00:00:00 2001 From: Brandur Date: Sat, 26 Jul 2025 20:16:16 -1000 Subject: [PATCH] Upgrade to golangci-lint v2.2.1 + disable some annoying lints + fix others Here, upgrade golangci-lint to v2.2.1, mostly because my local version is upgraded and throwing problems. Some really, really, really-no-good lints have been added (`noinlineerr` and `wsl_v5`) which we disable. Some others have been added which are good, which we keep. One checks that there's a consistent whitespace between embedded structs and other properties on a struct definition. Another makes sure we're using the context form of logging functions, so `WarnContext` instead of `Warn`. --- .github/workflows/ci.yaml | 2 +- .golangci.yaml | 2 ++ client.go | 8 +++---- client_test.go | 13 ++++++++--- example_complete_job_within_tx_test.go | 1 + example_graceful_shutdown_test.go | 1 + example_job_cancel_from_client_test.go | 1 + example_queue_pause_test.go | 1 + internal/dbunique/db_unique_test.go | 9 ++++++++ internal/jobcompleter/job_completer.go | 6 ++--- internal/jobcompleter/job_completer_test.go | 2 ++ internal/jobexecutor/job_executor.go | 4 ++-- internal/jobexecutor/job_executor_test.go | 2 +- internal/leadership/elector.go | 2 +- internal/maintenance/job_cleaner.go | 2 +- internal/maintenance/queue_maintainer.go | 1 + .../riverinternaltest/sharedtx/shared_tx.go | 3 +++ periodic_job_test.go | 1 + plugin_test.go | 2 ++ producer.go | 22 +++++++++---------- producer_test.go | 1 + .../river_database_sql_driver.go | 2 ++ riverdriver/riverpgxv5/river_pgx_v5_driver.go | 1 + .../riversqlite/river_sqlite_driver.go | 10 +++++---- riverlog/river_log.go | 1 + .../slogtest/slog_test_handler_test.go | 14 +++++++----- rivershared/startstop/start_stop.go | 1 + .../startstoptest/startstoptest_test.go | 1 + worker.go | 1 + 29 files changed, 81 insertions(+), 36 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 3ad5ab84..964368a1 100644 --- a/.github/workflows/ci.yaml +++ b/.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. diff --git a/.golangci.yaml b/.golangci.yaml index 389a446e..d324d749 100644 --- a/.golangci.yaml +++ b/.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: diff --git a/client.go b/client.go index c5710b8a..67ce29d9 100644 --- a/client.go +++ b/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 `.` 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{ diff --git a/client_test.go b/client_test.go index 89c385cd..486000a5 100644 --- a/client_test.go +++ b/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), ¬if.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"` } diff --git a/example_complete_job_within_tx_test.go b/example_complete_job_within_tx_test.go index 67b3b213..d9318200 100644 --- a/example_complete_job_within_tx_test.go +++ b/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 } diff --git a/example_graceful_shutdown_test.go b/example_graceful_shutdown_test.go index f9e77983..eb512009 100644 --- a/example_graceful_shutdown_test.go +++ b/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{} } diff --git a/example_job_cancel_from_client_test.go b/example_job_cancel_from_client_test.go index 4aa43d0a..5b0c7be6 100644 --- a/example_job_cancel_from_client_test.go +++ b/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 } diff --git a/example_queue_pause_test.go b/example_queue_pause_test.go index 22bf216b..ebefd32c 100644 --- a/example_queue_pause_test.go +++ b/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 } diff --git a/internal/dbunique/db_unique_test.go b/internal/dbunique/db_unique_test.go index 56c0e35d..99838a46 100644 --- a/internal/dbunique/db_unique_test.go +++ b/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{ diff --git a/internal/jobcompleter/job_completer.go b/internal/jobcompleter/job_completer.go index b6930723..c74b09b6 100644 --- a/internal/jobcompleter/job_completer.go +++ b/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 diff --git a/internal/jobcompleter/job_completer_test.go b/internal/jobcompleter/job_completer_test.go index 09451bcb..397279af 100644 --- a/internal/jobcompleter/job_completer_test.go +++ b/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 } diff --git a/internal/jobexecutor/job_executor.go b/internal/jobexecutor/job_executor.go index af9438ec..a550e147 100644 --- a/internal/jobexecutor/job_executor.go +++ b/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) } diff --git a/internal/jobexecutor/job_executor_test.go b/internal/jobexecutor/job_executor_test.go index 6f70db01..40bc9111 100644 --- a/internal/jobexecutor/job_executor_test.go +++ b/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) }() diff --git a/internal/leadership/elector.go b/internal/leadership/elector.go index 39260c0a..8d3592ed 100644 --- a/internal/leadership/elector.go +++ b/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 } diff --git a/internal/maintenance/job_cleaner.go b/internal/maintenance/job_cleaner.go index 8932625f..ec5f1876 100644 --- a/internal/maintenance/job_cleaner.go +++ b/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 diff --git a/internal/maintenance/queue_maintainer.go b/internal/maintenance/queue_maintainer.go index de7171bc..7f980623 100644 --- a/internal/maintenance/queue_maintainer.go +++ b/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 } diff --git a/internal/riverinternaltest/sharedtx/shared_tx.go b/internal/riverinternaltest/sharedtx/shared_tx.go index a9f3a086..74e14ec8 100644 --- a/internal/riverinternaltest/sharedtx/shared_tx.go +++ b/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 } diff --git a/periodic_job_test.go b/periodic_job_test.go index c8a4bc5e..ffe88743 100644 --- a/periodic_job_test.go +++ b/periodic_job_test.go @@ -54,6 +54,7 @@ func TestPeriodicJobBundle(t *testing.T) { type TestJobArgs struct { testutil.JobArgsReflectKind[TestJobArgs] + JobNum int `json:"job_num"` } diff --git a/plugin_test.go b/plugin_test.go index 7943f146..13135c6f 100644 --- a/plugin_test.go +++ b/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 } diff --git a/producer.go b/producer.go index 0bd1d680..3cf1fb8c 100644 --- a/producer.go +++ b/producer.go @@ -260,9 +260,9 @@ func (p *producer) Start(ctx context.Context) error { } func (p *producer) Stop() { - p.Logger.Debug(p.Name+": Stopping", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) + p.Logger.Debug(p.Name+": Stopping", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) //nolint:noctx p.BaseStartStop.Stop() - p.Logger.Debug(p.Name+": Stop returned", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) + p.Logger.Debug(p.Name+": Stop returned", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) //nolint:noctx } // Start starts the producer. It backgrounds a goroutine which is stopped when @@ -414,13 +414,13 @@ func (p *producer) StartWorkContext(fetchCtx, workCtx context.Context) error { } p.fetchAndRunLoop(fetchCtx, workCtx) - p.Logger.Debug(p.Name+": Entering shutdown loop", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) + p.Logger.DebugContext(workCtx, p.Name+": Entering shutdown loop", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) p.executorShutdownLoop() - p.Logger.Debug(p.Name+": Shutdown loop exited, awaiting subroutines", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) + p.Logger.DebugContext(workCtx, p.Name+": Shutdown loop exited, awaiting subroutines", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) cancelSubroutines(fmt.Errorf("producer stopped: %w", startstop.ErrStop)) subroutineWG.Wait() - p.Logger.Debug(p.Name+": Shutdown subroutines completed, finalizing", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) + p.Logger.DebugContext(workCtx, p.Name+": Shutdown subroutines completed, finalizing", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) p.finalizeShutdown(context.WithoutCancel(fetchCtx)) }() @@ -564,7 +564,7 @@ func (p *producer) fetchAndRunLoop(fetchCtx, workCtx context.Context) { p.Logger.DebugContext(workCtx, p.Name+": Unknown queue control action", "action", msg.Action) } case jobID := <-p.cancelCh: - p.maybeCancelJob(jobID) + p.maybeCancelJob(workCtx, jobID) case <-p.fetchLimiter.C(): p.innerFetchLoop(workCtx, fetchResultCh) // Ensure we can't start another fetch when fetchCtx is done, even if @@ -647,7 +647,7 @@ func (p *producer) innerFetchLoop(workCtx context.Context, fetchResultCh chan pr case result := <-p.jobResultCh: p.removeActiveJob(result) case jobID := <-p.cancelCh: - p.maybeCancelJob(jobID) + p.maybeCancelJob(workCtx, jobID) } } } @@ -662,7 +662,7 @@ func (p *producer) executorShutdownLoop() { } func (p *producer) finalizeShutdown(ctx context.Context) { - p.Logger.Debug(p.Name + ": Finalizing shutdown") + p.Logger.DebugContext(ctx, p.Name+": Finalizing shutdown") const ( maxAttempts = 4 // Maximum number of shutdown attempts @@ -724,12 +724,12 @@ func (p *producer) removeActiveJob(job *rivertype.JobRow) { p.state.JobFinish(job) } -func (p *producer) maybeCancelJob(id int64) { +func (p *producer) maybeCancelJob(ctx context.Context, id int64) { executor, ok := p.activeJobs[id] if !ok { return } - executor.Cancel() + executor.Cancel(ctx) } func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultCh chan<- producerFetchResult) { @@ -756,7 +756,7 @@ func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultC Schema: p.config.Schema, }) if err != nil { - p.Logger.Error(p.Name+": Error fetching jobs", slog.String("err", err.Error()), slog.String("queue", p.config.Queue)) + p.Logger.ErrorContext(ctx, p.Name+": Error fetching jobs", slog.String("err", err.Error()), slog.String("queue", p.config.Queue)) fetchResultCh <- producerFetchResult{err: err} return } diff --git a/producer_test.go b/producer_test.go index 87895f33..538bf272 100644 --- a/producer_test.go +++ b/producer_test.go @@ -73,6 +73,7 @@ func Test_Producer_CanSafelyCompleteJobsWhileFetchingNewOnes(t *testing.T) { type WithJobNumArgs struct { testutil.JobArgsReflectKind[WithJobNumArgs] + JobNum int `json:"job_num"` } diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go index 8484973d..3b673a10 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/riverdriver/riverdatabasesql/river_database_sql_driver.go @@ -986,6 +986,7 @@ func (e *Executor) TableTruncate(ctx context.Context, params *riverdriver.TableT type ExecutorTx struct { Executor + tx *sql.Tx } @@ -1010,6 +1011,7 @@ func (t *ExecutorTx) Rollback(ctx context.Context) error { type ExecutorSubTx struct { Executor + beginOnce *savepointutil.BeginOnlyOnce savepointNum int tx *sql.Tx diff --git a/riverdriver/riverpgxv5/river_pgx_v5_driver.go b/riverdriver/riverpgxv5/river_pgx_v5_driver.go index 4dff6d81..499acc72 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/riverdriver/riverpgxv5/river_pgx_v5_driver.go @@ -975,6 +975,7 @@ func (e *Executor) TableTruncate(ctx context.Context, params *riverdriver.TableT type ExecutorTx struct { Executor + tx pgx.Tx } diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index 27731b50..2469d408 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -382,7 +382,7 @@ func (e *Executor) JobDeleteMany(ctx context.Context, params *riverdriver.JobDel // sqlc is buggy and can't parse it. // // The SQL appends a new `attempted_by` to a job's `attempted_by` array (a jsonb -// array), which in SQLite is really quite difficult to to because there aren't +// array), which in SQLite is really quite difficult to because there aren't // any arrays or array functions. I tried every version of this in sqlc I could // come up with, but there was a bug in every direction that blocked it. e.g. // @@ -392,7 +392,7 @@ func (e *Executor) JobDeleteMany(ctx context.Context, params *riverdriver.JobDel // - I tried putting it in a CTE, but this is paired with an `UPDATE` statement, // and `UPDATE ... FROM` isn't support in sqlc for SQLite. // -// I'm really hoping this could be fixed one day by by bringing it back in with +// I'm really hoping this could be fixed one day by bringing it back in with // the rest of the job definitions, but it'll require some sqlc fixes for that // to work. Frustratingly, some of these fixes actually exist already [1], but // just can't be merged/release due to bottlenecks in the sqlc project. @@ -732,7 +732,7 @@ func (e *Executor) JobKindListByPrefix(ctx context.Context, params *riverdriver. kinds, err := dbsqlc.New().JobKindListByPrefix(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobKindListByPrefixParams{ After: params.After, Exclude: exclude, - Max: int64(min(params.Max, math.MaxInt32)), //nolint:gosec + Max: int64(min(params.Max, math.MaxInt32)), Prefix: params.Prefix, }) if err != nil { @@ -1259,7 +1259,7 @@ func (e *Executor) QueueNameListByPrefix(ctx context.Context, params *riverdrive queueNames, err := dbsqlc.New().QueueNameListByPrefix(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.QueueNameListByPrefixParams{ After: params.After, Exclude: exclude, - Max: int64(min(params.Max, math.MaxInt32)), //nolint:gosec + Max: int64(min(params.Max, math.MaxInt32)), Prefix: params.Prefix, }) if err != nil { @@ -1405,6 +1405,7 @@ func (e *Executor) TableTruncate(ctx context.Context, params *riverdriver.TableT type ExecutorTx struct { Executor + tx *sql.Tx } @@ -1430,6 +1431,7 @@ func (t *ExecutorTx) Rollback(ctx context.Context) error { type ExecutorSubTx struct { Executor + beginOnce *savepointutil.BeginOnlyOnce savepointNum int tx *sql.Tx diff --git a/riverlog/river_log.go b/riverlog/river_log.go index c8b9a892..5f689858 100644 --- a/riverlog/river_log.go +++ b/riverlog/river_log.go @@ -42,6 +42,7 @@ func Logger(ctx context.Context) *slog.Logger { type Middleware struct { baseservice.BaseService rivertype.Middleware + config *MiddlewareConfig newCustomContext func(ctx context.Context, w io.Writer) context.Context newSlogHandler func(w io.Writer) slog.Handler diff --git a/rivershared/slogtest/slog_test_handler_test.go b/rivershared/slogtest/slog_test_handler_test.go index 5f15fb03..28aac090 100644 --- a/rivershared/slogtest/slog_test_handler_test.go +++ b/rivershared/slogtest/slog_test_handler_test.go @@ -1,6 +1,7 @@ package slogtest import ( + "context" "log/slog" "sync" "testing" @@ -12,6 +13,8 @@ import ( func TestSlogTestHandler_levels(t *testing.T) { t.Parallel() + ctx := context.Background() + testCases := []struct { desc string level slog.Level @@ -27,10 +30,10 @@ func TestSlogTestHandler_levels(t *testing.T) { logger := NewLogger(t, &slog.HandlerOptions{Level: tt.level}) - logger.Debug("debug message") - logger.Info("info message") - logger.Warn("warn message") - logger.Error("error message") + logger.DebugContext(ctx, "debug message") + logger.InfoContext(ctx, "info message") + logger.WarnContext(ctx, "warn message") + logger.ErrorContext(ctx, "error message") }) } } @@ -39,6 +42,7 @@ func TestSlogTestHandler_stress(t *testing.T) { t.Parallel() var ( + ctx = context.Background() logger = NewLogger(t, nil) wg sync.WaitGroup ) @@ -47,7 +51,7 @@ func TestSlogTestHandler_stress(t *testing.T) { wg.Add(1) go func() { for range 100 { - logger.Info("message", "key", "value") + logger.InfoContext(ctx, "message", "key", "value") } wg.Done() }() diff --git a/rivershared/startstop/start_stop.go b/rivershared/startstop/start_stop.go index ed252435..59e26922 100644 --- a/rivershared/startstop/start_stop.go +++ b/rivershared/startstop/start_stop.go @@ -243,6 +243,7 @@ func (s *BaseStartStop) StoppedUnsafe() <-chan struct{} { return s.stopped } type startStopFunc struct { BaseStartStop + startFunc func(ctx context.Context, shouldStart bool, started, stopped func()) error } diff --git a/rivershared/startstoptest/startstoptest_test.go b/rivershared/startstoptest/startstoptest_test.go index b3cef024..3515bd7f 100644 --- a/rivershared/startstoptest/startstoptest_test.go +++ b/rivershared/startstoptest/startstoptest_test.go @@ -15,6 +15,7 @@ import ( type MyService struct { startstop.BaseStartStop + logger *slog.Logger startErr error } diff --git a/worker.go b/worker.go index b5cac4c2..1a5aedf6 100644 --- a/worker.go +++ b/worker.go @@ -181,6 +181,7 @@ func (w Workers) add(jobArgs JobArgs, workUnitFactory workunit.WorkUnitFactory) // workFunc implements JobArgs and is used to wrap a function given to WorkFunc. type workFunc[T JobArgs] struct { WorkerDefaults[T] + kind string f func(context.Context, *Job[T]) error }