From 90bdd5623e6d6798859d07c09757f59fd397e9ba Mon Sep 17 00:00:00 2001 From: Brandur Date: Tue, 3 Nov 2026 23:12:50 -0600 Subject: [PATCH 1/2] [go, js, rust] Fix context canceled problem in start when stopping Fixes #1576. When the enqueuer was still starting up, an immediate stop could cancel its maintenance context while it was in staggered sleep. The sleep would return immeditaely, but the startup continued and tried to insert `RunOnStart` jobs which would error as it tried to start a transaction on an already-cancelled context. Fix is to do more cancelled context checking. I had Codex go through and audit the other maintenance services and ports for similar problem. It found a few more including some in JS and Rust, which are also fixed. Java and Ruby were unaffected. --- CHANGELOG.md | 1 + internal/maintenance/job_cleaner.go | 11 +- internal/maintenance/job_rescuer.go | 11 +- internal/maintenance/job_scheduler.go | 8 +- internal/maintenance/periodic_job_enqueuer.go | 42 +++- .../periodic_job_enqueuer_shutdown_test.go | 190 ++++++++++++++++ internal/maintenance/queue_cleaner.go | 11 +- .../maintenance/queue_maintainer_leader.go | 4 + internal/maintenance/reindexer.go | 16 +- internal/maintenance/shutdown_test.go | 207 ++++++++++++++++++ .../sqlite_notification_cleaner.go | 11 +- js/CHANGELOG.md | 1 + js/src/services.test.ts | 80 +++++++ js/src/services.ts | 4 +- rust/CHANGELOG.md | 1 + rust/riverqueue/src/maintenance/postgres.rs | 63 +++++- 16 files changed, 637 insertions(+), 24 deletions(-) create mode 100644 internal/maintenance/periodic_job_enqueuer_shutdown_test.go create mode 100644 internal/maintenance/shutdown_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index c7766e359..72fc73a1a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed a client ignoring queue pause, resume, or metadata changes until it restarted when the queue control notification was lost, as happens if it's sent while the client's notifier is reconnecting. Clients with a notifier now also check their queues' settings once a minute. [Issue #1579](https://github.com/riverqueue/river/issues/1579). - Fixed the notifier never reconnecting when its connection went silent without an error, like after a dropped network path, because its health check ping had no timeout. The ping now times out after 10 seconds. [Issue #1579](https://github.com/riverqueue/river/issues/1579). +- Fixed maintenance services logging errors during normal shutdown or leadership loss, including `PeriodicJobEnqueuer` reporting `Error starting transaction: context canceled` when stopped during staggered startup. Genuine query timeouts and database failures are still logged, and periodic job transactions now roll back even after cancellation. [PR #1577](https://github.com/riverqueue/river/pull/1577). ## [0.51.0] - 2026-10-31 diff --git a/internal/maintenance/job_cleaner.go b/internal/maintenance/job_cleaner.go index 185dc53cf..ed7e29848 100644 --- a/internal/maintenance/job_cleaner.go +++ b/internal/maintenance/job_cleaner.go @@ -146,6 +146,12 @@ func (s *JobCleaner) Start(ctx context.Context) error { //nolint:dupl } s.StaggerStart(ctx) + select { + case <-ctx.Done(): + stopped() + return nil + default: + } go func() { started() @@ -164,9 +170,10 @@ func (s *JobCleaner) Start(ctx context.Context) error { //nolint:dupl res, err := s.runOnce(ctx) if err != nil { - if !errors.Is(err, context.Canceled) { - s.Logger.ErrorContext(ctx, s.Name+": Error cleaning jobs", slog.String("error", err.Error())) + if ctx.Err() != nil { + return } + s.Logger.ErrorContext(ctx, s.Name+": Error cleaning jobs", slog.String("error", err.Error())) continue } diff --git a/internal/maintenance/job_rescuer.go b/internal/maintenance/job_rescuer.go index f35ae51ac..a00269451 100644 --- a/internal/maintenance/job_rescuer.go +++ b/internal/maintenance/job_rescuer.go @@ -143,6 +143,12 @@ func (s *JobRescuer) Start(ctx context.Context) error { } s.StaggerStart(ctx) + select { + case <-ctx.Done(): + stopped() + return nil + default: + } go func() { started() @@ -161,9 +167,10 @@ func (s *JobRescuer) Start(ctx context.Context) error { res, err := s.runOnce(ctx) if err != nil { - if !errors.Is(err, context.Canceled) { - s.Logger.ErrorContext(ctx, s.Name+": Error rescuing jobs", slog.String("error", err.Error())) + if ctx.Err() != nil { + return } + s.Logger.ErrorContext(ctx, s.Name+": Error rescuing jobs", slog.String("error", err.Error())) continue } diff --git a/internal/maintenance/job_scheduler.go b/internal/maintenance/job_scheduler.go index 997f5d83c..d099ff067 100644 --- a/internal/maintenance/job_scheduler.go +++ b/internal/maintenance/job_scheduler.go @@ -107,13 +107,19 @@ func NewJobScheduler(archetype *baseservice.Archetype, config *JobSchedulerConfi }) } -func (s *JobScheduler) Start(ctx context.Context) error { +func (s *JobScheduler) Start(ctx context.Context) error { //nolint:dupl ctx, shouldStart, started, stopped := s.StartInit(ctx) if !shouldStart { return nil } s.StaggerStart(ctx) + select { + case <-ctx.Done(): + stopped() + return nil + default: + } go func() { started() diff --git a/internal/maintenance/periodic_job_enqueuer.go b/internal/maintenance/periodic_job_enqueuer.go index aa67a4695..ea6aa1e2c 100644 --- a/internal/maintenance/periodic_job_enqueuer.go +++ b/internal/maintenance/periodic_job_enqueuer.go @@ -18,6 +18,7 @@ import ( "github.com/riverqueue/river/rivershared/riversharedmaintenance" "github.com/riverqueue/river/rivershared/startstop" "github.com/riverqueue/river/rivershared/testsignal" + "github.com/riverqueue/river/rivershared/util/dbutil" "github.com/riverqueue/river/rivershared/util/maputil" "github.com/riverqueue/river/rivershared/util/sliceutil" "github.com/riverqueue/river/rivershared/util/testutil" @@ -320,6 +321,12 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { } s.StaggerStart(ctx) + select { + case <-ctx.Done(): + stopped() + return nil + default: + } var ( initialPeriodicJobs []*riverpilot.PeriodicJob @@ -390,6 +397,10 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { var lastHandleSeen rivertype.PeriodicJobHandle = -1 // so handle 0 is considered validateInsertRunOnStartAndScheduleNewlyAdded := func() { + if ctx.Err() != nil { + return + } + var ( insertParamsMany []*rivertype.JobInsertParams now = s.Time.Now() @@ -532,38 +543,49 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { } func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany []*rivertype.JobInsertParams, periodicJobUpsertParams *riverpilot.PeriodicJobUpsertManyParams, periodicJobsToKeepAlive map[rivertype.PeriodicJobHandle]*PeriodicJob) { - if len(insertParamsMany) < 1 && len(periodicJobUpsertParams.Jobs) < 1 { + if ctx.Err() != nil || (len(insertParamsMany) < 1 && len(periodicJobUpsertParams.Jobs) < 1) { return } + // Check the service context when logging, so query timeouts still surface + // while shutdown errors (including driver-specific errors) stay quiet. + serviceCtx := ctx ctx, cancel := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault) defer cancel() tx, err := s.exec.Begin(ctx) if err != nil { - s.Logger.ErrorContext(ctx, s.Name+": Error starting transaction", "error", err.Error()) + if serviceCtx.Err() == nil { + s.Logger.ErrorContext(ctx, s.Name+": Error starting transaction", "error", err.Error()) + } return } - defer tx.Rollback(ctx) + defer dbutil.RollbackWithoutCancel(ctx, tx) if len(insertParamsMany) > 0 { if _, err := s.Config.Insert(ctx, tx, insertParamsMany); err != nil { - s.Logger.ErrorContext(ctx, s.Name+": Error inserting periodic jobs", - "error", err.Error(), "num_jobs", len(insertParamsMany)) + if serviceCtx.Err() == nil { + s.Logger.ErrorContext(ctx, s.Name+": Error inserting periodic jobs", + "error", err.Error(), "num_jobs", len(insertParamsMany)) + } return } } if len(periodicJobUpsertParams.Jobs) > 0 { if _, err = s.Config.Pilot.PeriodicJobUpsertMany(ctx, tx, periodicJobUpsertParams); err != nil { - s.Logger.ErrorContext(ctx, s.Name+": Error upserting periodic job next run times", - "error", err.Error(), "num_jobs", len(insertParamsMany), "num_next_run_at_upserts", len(periodicJobUpsertParams.Jobs)) + if serviceCtx.Err() == nil { + s.Logger.ErrorContext(ctx, s.Name+": Error upserting periodic job next run times", + "error", err.Error(), "num_jobs", len(insertParamsMany), "num_next_run_at_upserts", len(periodicJobUpsertParams.Jobs)) + } return } } if err := tx.Commit(ctx); err != nil { - s.Logger.ErrorContext(ctx, s.Name+": Error committing transaction", "error", err.Error()) + if serviceCtx.Err() == nil { + s.Logger.ErrorContext(ctx, s.Name+": Error committing transaction", "error", err.Error()) + } return } @@ -649,7 +671,9 @@ func (s *PeriodicJobEnqueuer) periodicJobKeepAliveAndReapPeriodically(ctx contex ID: ids, Schema: s.Config.Schema, }); err != nil { - s.Logger.ErrorContext(ctx, s.Name+": Error executing periodic job keep alive and reap", "error", err.Error()) + if ctx.Err() == nil { + s.Logger.ErrorContext(ctx, s.Name+": Error executing periodic job keep alive and reap", "error", err.Error()) + } return } } diff --git a/internal/maintenance/periodic_job_enqueuer_shutdown_test.go b/internal/maintenance/periodic_job_enqueuer_shutdown_test.go new file mode 100644 index 000000000..7d26bba11 --- /dev/null +++ b/internal/maintenance/periodic_job_enqueuer_shutdown_test.go @@ -0,0 +1,190 @@ +package maintenance + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "os" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" + + "github.com/riverqueue/river/riverdriver" + "github.com/riverqueue/river/rivershared/riverpilot" + "github.com/riverqueue/river/rivershared/riversharedtest" + "github.com/riverqueue/river/rivershared/testsignal" + "github.com/riverqueue/river/rivertype" +) + +type periodicShutdownExecutor struct { + riverdriver.Executor + + beginFunc func(context.Context) error + tx *periodicShutdownTx +} + +func (e *periodicShutdownExecutor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) { + if err := e.beginFunc(ctx); err != nil { + return nil, err + } + return e.tx, nil +} + +type periodicShutdownTx struct { + riverdriver.ExecutorTx + + commitFunc func(context.Context) error + rollbackFunc func(context.Context) error +} + +func (tx *periodicShutdownTx) Commit(ctx context.Context) error { return tx.commitFunc(ctx) } + +func (tx *periodicShutdownTx) Rollback(ctx context.Context) error { return tx.rollbackFunc(ctx) } + +func TestPeriodicJobEnqueuer_ShutdownDuringStagger(t *testing.T) { + t.Parallel() + + for _, stopService := range []bool{false, true} { + t.Run(fmt.Sprintf("StopService=%t", stopService), func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + var logOutput bytes.Buffer + archetype := riversharedtest.BaseServiceArchetype(t) + archetype.Logger = slog.New(slog.NewTextHandler(&logOutput, &slog.HandlerOptions{Level: slog.LevelError})) + var beginCalls, constructorCalls, getAllCalls int + pilot := NewPilotPeriodicJobMock() + pilot.PeriodicJobGetAllMock = func(context.Context, riverdriver.Executor, *riverpilot.PeriodicJobGetAllParams) ([]*riverpilot.PeriodicJob, error) { + getAllCalls++ + return nil, nil + } + svc, err := NewPeriodicJobEnqueuer(archetype, &PeriodicJobEnqueuerConfig{ + PeriodicJobs: []*PeriodicJob{{ + ConstructorFunc: func() (*rivertype.JobInsertParams, error) { + constructorCalls++ + return &rivertype.JobInsertParams{}, nil + }, + RunOnStart: true, + ScheduleFunc: func(now time.Time) time.Time { return now.Add(time.Hour) }, + }}, + Pilot: pilot, + }, &periodicShutdownExecutor{beginFunc: func(ctx context.Context) error { + beginCalls++ + return ctx.Err() + }}) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var startResult testsignal.TestSignal[error] + startResult.Init(t) + go func() { startResult.Signal(svc.Start(ctx)) }() + // Wait until Start is sleeping in the real stagger, without + // depending on its randomly selected duration. + synctest.Wait() + if stopService { + svc.Stop() + } else { + cancel() + } + err = startResult.WaitOrTimeout() + svc.Stop() + require.Empty(t, logOutput.String()) + require.NoError(t, err) + require.Zero(t, beginCalls) + require.Zero(t, constructorCalls) + require.Zero(t, getAllCalls) + }) + }) + } +} + +func TestPeriodicJobEnqueuer_TransactionErrorLogging(t *testing.T) { + t.Parallel() + + for _, operation := range []string{"Begin", "Commit", "Insert", "Upsert"} { + t.Run(operation, func(t *testing.T) { + t.Parallel() + + for _, outcome := range []string{"Canceled", "DatabaseError", "DriverErrorOnCancellation", "LocalTimeout", "ParentDeadline"} { + t.Run(outcome, func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + var logOutput bytes.Buffer + archetype := riversharedtest.BaseServiceArchetype(t) + archetype.Logger = slog.New(slog.NewTextHandler(&logOutput, &slog.HandlerOptions{Level: slog.LevelError})) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if outcome == "ParentDeadline" { + var cancelDeadline context.CancelFunc + ctx, cancelDeadline = context.WithTimeout(ctx, time.Second) + defer cancelDeadline() + } + dbErr := errors.New("database failed") + operationFunc := func(queryCtx context.Context, name string) error { + if name != operation { + return nil + } + switch outcome { + case "Canceled": + cancel() + return fmt.Errorf("query interrupted: %w", queryCtx.Err()) + case "DriverErrorOnCancellation": + cancel() + return fmt.Errorf("write failed: %w", os.ErrDeadlineExceeded) + case "LocalTimeout", "ParentDeadline": + <-queryCtx.Done() + return queryCtx.Err() + default: + return dbErr + } + } + rolledBack := false + exec := &periodicShutdownExecutor{ + beginFunc: func(ctx context.Context) error { return operationFunc(ctx, "Begin") }, + tx: &periodicShutdownTx{ + commitFunc: func(ctx context.Context) error { return operationFunc(ctx, "Commit") }, + rollbackFunc: func(ctx context.Context) error { + rolledBack = true + require.NoError(t, ctx.Err(), "rollback must survive query cancellation") + return nil + }, + }, + } + pilot := NewPilotPeriodicJobMock() + pilot.PeriodicJobUpsertManyMock = func(ctx context.Context, _ riverdriver.Executor, _ *riverpilot.PeriodicJobUpsertManyParams) ([]*riverpilot.PeriodicJob, error) { + return nil, operationFunc(ctx, "Upsert") + } + svc, err := NewPeriodicJobEnqueuer(archetype, &PeriodicJobEnqueuerConfig{ + Insert: func(ctx context.Context, _ riverdriver.ExecutorTx, _ []*rivertype.JobInsertParams) ([]*rivertype.JobInsertResult, error) { + return nil, operationFunc(ctx, "Insert") + }, + Pilot: pilot, + }, exec) + require.NoError(t, err) + svc.TestSignals.Init(t) + svc.insertBatch(ctx, []*rivertype.JobInsertParams{{}}, &riverpilot.PeriodicJobUpsertManyParams{ + Jobs: []*riverpilot.PeriodicJobUpsertParams{{ID: "periodic"}}, + }, nil) + switch outcome { + case "DatabaseError": + require.Contains(t, logOutput.String(), dbErr.Error()) + case "LocalTimeout": + require.Contains(t, logOutput.String(), context.DeadlineExceeded.Error()) + default: + require.Empty(t, logOutput.String()) + } + require.Equal(t, operation != "Begin", rolledBack) + svc.TestSignals.InsertedJobs.RequireEmpty() + svc.TestSignals.PeriodicJobUpserted.RequireEmpty() + }) + }) + } + }) + } +} diff --git a/internal/maintenance/queue_cleaner.go b/internal/maintenance/queue_cleaner.go index e417558b9..b81d06d53 100644 --- a/internal/maintenance/queue_cleaner.go +++ b/internal/maintenance/queue_cleaner.go @@ -107,6 +107,12 @@ func (s *QueueCleaner) Start(ctx context.Context) error { } s.StaggerStart(ctx) + select { + case <-ctx.Done(): + stopped() + return nil + default: + } go func() { started() @@ -125,9 +131,10 @@ func (s *QueueCleaner) Start(ctx context.Context) error { res, err := s.runOnce(ctx) if err != nil { - if !errors.Is(err, context.Canceled) { - s.Logger.ErrorContext(ctx, s.Name+": Error cleaning queues", slog.String("error", err.Error())) + if ctx.Err() != nil { + return } + s.Logger.ErrorContext(ctx, s.Name+": Error cleaning queues", slog.String("error", err.Error())) continue } diff --git a/internal/maintenance/queue_maintainer_leader.go b/internal/maintenance/queue_maintainer_leader.go index 1cc6add13..bf166eefa 100644 --- a/internal/maintenance/queue_maintainer_leader.go +++ b/internal/maintenance/queue_maintainer_leader.go @@ -146,6 +146,10 @@ func (s *QueueMaintainerLeader) tryStart(ctx context.Context, epoch int64, term return } + if ctx.Err() != nil { + return + } + s.Logger.ErrorContext(ctx, s.Name+": Error starting queue maintainer", slog.String("err", lastErr.Error()), slog.Int("attempt", attempt)) diff --git a/internal/maintenance/reindexer.go b/internal/maintenance/reindexer.go index cc167164e..1a6bcebdf 100644 --- a/internal/maintenance/reindexer.go +++ b/internal/maintenance/reindexer.go @@ -114,6 +114,12 @@ func (s *Reindexer) Start(ctx context.Context) error { } s.StaggerStart(ctx) + select { + case <-ctx.Done(): + stopped() + return nil + default: + } go func() { started() @@ -140,18 +146,20 @@ func (s *Reindexer) Start(ctx context.Context) error { case <-timerUntilNextRun.C: reindexableIndexNames, err := s.reindexableIndexNames(ctx) if err != nil { - if !errors.Is(err, context.Canceled) { - s.Logger.ErrorContext(ctx, s.Name+": Error listing reindexable indexes", slog.String("error", err.Error())) + if ctx.Err() != nil { + return } + s.Logger.ErrorContext(ctx, s.Name+": Error listing reindexable indexes", slog.String("error", err.Error())) scheduleNextRun() continue } for _, indexName := range reindexableIndexNames { if _, err := s.reindexOne(ctx, indexName); err != nil { - if !errors.Is(err, context.Canceled) { - s.Logger.ErrorContext(ctx, s.Name+": Error reindexing", slog.String("error", err.Error()), slog.String("index_name", indexName)) + if ctx.Err() != nil { + return } + s.Logger.ErrorContext(ctx, s.Name+": Error reindexing", slog.String("error", err.Error()), slog.String("index_name", indexName)) continue } } diff --git a/internal/maintenance/shutdown_test.go b/internal/maintenance/shutdown_test.go new file mode 100644 index 000000000..4a51ca755 --- /dev/null +++ b/internal/maintenance/shutdown_test.go @@ -0,0 +1,207 @@ +package maintenance + +import ( + "bytes" + "context" + "fmt" + "log/slog" + "os" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" + + "github.com/riverqueue/river/internal/workunit" + "github.com/riverqueue/river/riverdriver" + "github.com/riverqueue/river/rivershared/baseservice" + "github.com/riverqueue/river/rivershared/riverpilot" + "github.com/riverqueue/river/rivershared/riversharedtest" + "github.com/riverqueue/river/rivershared/startstop" + "github.com/riverqueue/river/rivershared/testsignal" + "github.com/riverqueue/river/rivertype" +) + +// Inject the different error shapes drivers can return when cancellation +// interrupts a query, independently of a database's cancellation timing. +type maintenanceShutdownExecutor struct { + riverdriver.Executor + + operationFunc func(context.Context, string) error +} + +func (e *maintenanceShutdownExecutor) IndexesExist(ctx context.Context, params *riverdriver.IndexesExistParams) (map[string]bool, error) { + return map[string]bool{"index": true}, e.operationFunc(ctx, "IndexesExist") +} + +func (e *maintenanceShutdownExecutor) IndexReindex(ctx context.Context, _ *riverdriver.IndexReindexParams) error { + return e.operationFunc(ctx, "IndexReindex") +} + +func (*maintenanceShutdownExecutor) IndexReindexArtifacts(context.Context, *riverdriver.IndexReindexArtifactsParams) ([]string, error) { + return nil, nil +} + +func (e *maintenanceShutdownExecutor) JobDeleteBefore(ctx context.Context, _ *riverdriver.JobDeleteBeforeParams) (int, error) { + return 0, e.operationFunc(ctx, "JobDeleteBefore") +} + +func (e *maintenanceShutdownExecutor) JobGetStuck(ctx context.Context, _ *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) { + return []*rivertype.JobRow{{ID: 1, Kind: "unknown", Metadata: []byte(`{}`)}}, e.operationFunc(ctx, "JobGetStuck") +} + +func (e *maintenanceShutdownExecutor) JobRescueMany(ctx context.Context, _ *riverdriver.JobRescueManyParams) (*struct{}, error) { + return nil, e.operationFunc(ctx, "JobRescueMany") +} + +func (e *maintenanceShutdownExecutor) NotificationDeleteBefore(ctx context.Context, _ *riverdriver.NotificationDeleteBeforeParams) (int, error) { + return 0, e.operationFunc(ctx, "NotificationDeleteBefore") +} + +func (e *maintenanceShutdownExecutor) QueueDeleteExpired(ctx context.Context, _ *riverdriver.QueueDeleteExpiredParams) ([]string, error) { + return nil, e.operationFunc(ctx, "QueueDeleteExpired") +} + +func TestMaintenance_ShutdownErrorLogging(t *testing.T) { + t.Parallel() + + type testBundle struct { + logOutput bytes.Buffer + queried testsignal.TestSignal[struct{}] + service startstop.Service + } + + setup := func(t *testing.T, operation string, queryFunc func(context.Context) error) *testBundle { + t.Helper() + + bundle := &testBundle{} + bundle.queried.Init(t) + archetype := riversharedtest.BaseServiceArchetype(t) + archetype.Logger = slog.New(slog.NewTextHandler(&bundle.logOutput, &slog.HandlerOptions{Level: slog.LevelError})) + exec := &maintenanceShutdownExecutor{operationFunc: func(ctx context.Context, name string) error { + if name != operation { + return nil + } + bundle.queried.Signal(struct{}{}) + return queryFunc(ctx) + }} + switch operation { + case "IndexesExist", "IndexReindex": + bundle.service = NewReindexer(archetype, &ReindexerConfig{ + IndexNames: []string{"index"}, + ScheduleFunc: func(now time.Time) time.Time { return now.Add(time.Second) }, + }, exec) + case "JobDeleteBefore": + bundle.service = NewJobCleaner(archetype, &JobCleanerConfig{}, exec) + case "JobGetStuck", "JobRescueMany": + bundle.service = NewRescuer(archetype, &JobRescuerConfig{ + ClientRetryPolicy: &SimpleClientRetryPolicy{}, + WorkUnitFactoryFunc: func(string) workunit.WorkUnitFactory { + return &callbackWorkUnitFactory{Callback: func(context.Context, *rivertype.JobRow) error { return nil }} + }, + }, exec) + case "NotificationDeleteBefore": + bundle.service = NewSQLiteNotificationCleaner(archetype, &SQLiteNotificationCleanerConfig{}, exec) + case "PeriodicJobGetAll", "PeriodicJobKeepAliveAndReap": + pilot := NewPilotPeriodicJobMock() + pilot.PeriodicJobGetAllMock = func(ctx context.Context, _ riverdriver.Executor, _ *riverpilot.PeriodicJobGetAllParams) ([]*riverpilot.PeriodicJob, error) { + return nil, exec.operationFunc(ctx, "PeriodicJobGetAll") + } + pilot.PeriodicJobKeepAliveAndReapMock = func(ctx context.Context, _ riverdriver.Executor, _ *riverpilot.PeriodicJobKeepAliveAndReapParams) ([]*riverpilot.PeriodicJob, error) { + return nil, exec.operationFunc(ctx, "PeriodicJobKeepAliveAndReap") + } + periodicExec := &periodicShutdownExecutor{ + beginFunc: func(context.Context) error { return nil }, + tx: &periodicShutdownTx{ + commitFunc: func(context.Context) error { return nil }, + rollbackFunc: func(context.Context) error { return nil }, + }, + } + enqueuer, err := NewPeriodicJobEnqueuer(archetype, &PeriodicJobEnqueuerConfig{ + PeriodicJobs: []*PeriodicJob{{ + ID: "periodic", + ConstructorFunc: func() (*rivertype.JobInsertParams, error) { return nil, ErrNoJobToInsert }, + ScheduleFunc: func(now time.Time) time.Time { return now.Add(time.Hour) }, + }}, + Pilot: pilot, + }, periodicExec) + require.NoError(t, err) + if operation == "PeriodicJobGetAll" { + enqueuer.StaggerStartupDisable(true) + bundle.service = maintenanceStartupService(archetype, enqueuer) + } else { + bundle.service = startstop.StartStopFunc(enqueuer.periodicJobKeepAliveAndReapPeriodically) + } + case "QueueDeleteExpired": + bundle.service = NewQueueCleaner(archetype, &QueueCleanerConfig{}, exec) + default: + t.Fatalf("unknown operation: %s", operation) + } + if svc, ok := bundle.service.(withStaggerStartupDisable); ok { + svc.StaggerStartupDisable(true) + } + t.Cleanup(bundle.service.Stop) + return bundle + } + + for _, operation := range []string{"IndexReindex", "IndexesExist", "JobDeleteBefore", "JobGetStuck", "JobRescueMany", "NotificationDeleteBefore", "PeriodicJobGetAll", "PeriodicJobKeepAliveAndReap", "QueueDeleteExpired"} { + t.Run(operation, func(t *testing.T) { + t.Parallel() + + for _, outcome := range []string{"ContextCanceled", "DatabaseError", "DeadlineExceeded", "Stop"} { + t.Run(outcome, func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + writeErr := fmt.Errorf("write failed: %w", os.ErrDeadlineExceeded) + bundle := setup(t, operation, func(ctx context.Context) error { + if outcome == "DatabaseError" { + return writeErr + } + if outcome == "DeadlineExceeded" { + return context.DeadlineExceeded + } + <-ctx.Done() + return writeErr + }) + require.NoError(t, bundle.service.Start(ctx)) + bundle.queried.WaitOrTimeout() + synctest.Wait() + if outcome == "ContextCanceled" { + cancel() + } + bundle.service.Stop() + switch outcome { + case "DatabaseError": + require.Contains(t, bundle.logOutput.String(), writeErr.Error()) + case "DeadlineExceeded": + require.Contains(t, bundle.logOutput.String(), context.DeadlineExceeded.Error()) + default: + require.Empty(t, bundle.logOutput.String()) + } + }) + }) + } + }) + } +} + +// Exercise synchronous startup failures through the leader's real retry loop. +func maintenanceStartupService(archetype *baseservice.Archetype, service startstop.Service) startstop.Service { + leader := NewQueueMaintainerLeader(archetype, &QueueMaintainerLeaderConfig{ + QueueMaintainer: NewQueueMaintainer(archetype, []startstop.Service{service}), + }) + return startstop.StartStopFunc(func(ctx context.Context, shouldStart bool, started, stopped func()) error { + if !shouldStart { + return nil + } + go func() { + started() + defer stopped() + leader.tryStart(ctx, 0, 0) + }() + return nil + }) +} diff --git a/internal/maintenance/sqlite_notification_cleaner.go b/internal/maintenance/sqlite_notification_cleaner.go index f5b6fbb59..232593dfa 100644 --- a/internal/maintenance/sqlite_notification_cleaner.go +++ b/internal/maintenance/sqlite_notification_cleaner.go @@ -108,6 +108,12 @@ func (s *SQLiteNotificationCleaner) Start(ctx context.Context) error { //nolint: } s.StaggerStart(ctx) + select { + case <-ctx.Done(): + stopped() + return nil + default: + } go func() { started() @@ -126,9 +132,10 @@ func (s *SQLiteNotificationCleaner) Start(ctx context.Context) error { //nolint: res, err := s.runOnce(ctx) if err != nil { - if !errors.Is(err, context.Canceled) { - s.Logger.ErrorContext(ctx, s.Name+": Error cleaning SQLite notifications", slog.String("error", err.Error())) + if ctx.Err() != nil { + return } + s.Logger.ErrorContext(ctx, s.Name+": Error cleaning SQLite notifications", slog.String("error", err.Error())) continue } diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index 124185cc4..7499b9c08 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed maintenance services logging failures during shutdown or leadership loss when a driver returned a socket or SQL error instead of the cancellation reason. Genuine query timeouts and database failures are still reported. [PR #1577](https://github.com/riverqueue/river/pull/1577). - Fixed a cancelled job that was retried being cancelled again when the client checked for cancellation requests, which it does while polling and after its notification stream reconnects. Retrying a job now clears its earlier attempt's cancellation request, like River for Go. [PR #1564](https://github.com/riverqueue/river/pull/1564). ## [0.3.0] - 2026-10-26 diff --git a/js/src/services.test.ts b/js/src/services.test.ts index 7c199d9c7..a2f1fb6e9 100644 --- a/js/src/services.test.ts +++ b/js/src/services.test.ts @@ -858,6 +858,86 @@ describe("RuntimeServices", () => { expect(calls).toBe(1); }); + it.each(["resign", "stop"] as const)( + "does not report driver-specific maintenance errors during %s", + async (shutdown) => { + const shared = new SharedLeadership(); + const events: RiverEvent[] = []; + const logged: string[] = []; + const signals: AbortSignal[] = []; + // A backend can reject with a socket or SQL error instead of the + // AbortSignal's reason when cancellation interrupts a query. + const interrupted = (signal: AbortSignal): Promise => { + signals.push(signal); + return new Promise((_resolve, reject) => { + signal.addEventListener( + "abort", + () => reject(new Error("connection closed during query")), + { once: true } + ); + }); + }; + const driver: RuntimeDriver = { + ...serviceDriver(shared), + maintenanceCleanJobs: (_leader, _params, _timeoutMs, signal) => + interrupted(signal), + maintenanceCleanNotifications: (_leader, _horizon, _limit, batch) => + interrupted(batch!.signal), + maintenanceCleanQueues: (_leader, _horizon, _limit, batch) => + interrupted(batch!.signal), + maintenanceGetStuck: (_leader, _horizon, _afterId, _limit, batch) => + interrupted(batch!.signal), + maintenanceReindex: (_leader, _indexes, _timeoutMs, signal) => + interrupted(signal), + maintenanceSchedule: (_leader, _params, batch) => + interrupted(batch!.signal), + }; + const services = new RuntimeServices({ + client: {} as Client, + clientId: "leader", + driver, + emit: (event) => { + events.push(event); + return Promise.resolve(); + }, + logger: { + error: (message) => logged.push(message), + warn: (message) => logged.push(message), + }, + maintenance: { + electionIntervalMs: 60_000, + jobCleanerIntervalMs: 60_000, + notificationCleanerIntervalMs: 60_000, + queueCleanerIntervalMs: 60_000, + reindexerSchedule: (after) => after.add({ milliseconds: 1 }), + rescuerIntervalMs: 60_000, + schedulerIntervalMs: 60_000, + }, + now: () => Temporal.Now.instant(), + periodicJobs: new PeriodicJobs(), + random: () => 0, + rescue: () => Promise.resolve(null), + }); + const controller = new AbortController(); + const running = services.run(controller.signal); + try { + await waitUntil(() => signals.length === 6); + if (shutdown === "resign") { + await expect(services.resignLeadership()).resolves.toBe(true); + } + } finally { + controller.abort(); + await running; + } + + expect(signals.every((signal) => signal.aborted)).toBe(true); + expect(logged).toEqual([]); + expect( + events.filter(({ kind }) => kind === "maintenance_failed") + ).toEqual([]); + } + ); + it("aborts backend work immediately when its exact term is resigned", async () => { const shared = new SharedLeadership(); const signals: AbortSignal[] = []; diff --git a/js/src/services.ts b/js/src/services.ts index 803d3bc95..b114ae496 100644 --- a/js/src/services.ts +++ b/js/src/services.ts @@ -591,7 +591,9 @@ export class RuntimeServices { }); } } catch (error: unknown) { - if (signal?.aborted === true && Object.is(error, signal.reason)) return; + // Drivers may report a socket or SQL error when cancellation interrupts + // a query. Check the service signal; batch timeouts must still be logged. + if (signal?.aborted === true) return; this.#lastError = errorMessage(error); // Maintenance retries on its own interval, which already bounds the // retry rate, exactly like River's other runtimes. diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index a1247d628..a24d5df34 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -14,6 +14,7 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. - Align migration CLI output with Go: `migrate-list` shows every bundled version and name, marking the latest applied version with `*`; migration progress uses the Go format and reports when nothing remains to apply; successful validation is silent. SQL output now requires `--show-sql`, including with `--dry-run`. [PR #1580](https://github.com/riverqueue/river/pull/1580). - Recover queue pause, resume, and metadata changes whose notifications were lost by polling persisted settings once a minute, including while the listener remains disconnected. [Issue #1579](https://github.com/riverqueue/river/issues/1579). +- Fixed Postgres maintenance, including reindexing, reporting a failure when a query error raced with shutdown or leadership loss. Successful query results, including commits, are preserved. [PR #1577](https://github.com/riverqueue/river/pull/1577). ## [0.5.0] - 2026-11-02 diff --git a/rust/riverqueue/src/maintenance/postgres.rs b/rust/riverqueue/src/maintenance/postgres.rs index 66357c0b4..3c910c13f 100644 --- a/rust/riverqueue/src/maintenance/postgres.rs +++ b/rust/riverqueue/src/maintenance/postgres.rs @@ -126,7 +126,7 @@ where None => std::future::pending().await, } }; - tokio::select! { + let result = tokio::select! { biased; result = &mut operation => result.map_err(MaintenanceError::from), () = cancel.cancelled() => { @@ -139,6 +139,14 @@ where let _ = tokio::time::timeout(CANCEL_GRACE, operation).await; Err(MaintenanceError::TimedOut) } + }; + // A completed query can report a driver error as cancellation arrives, + // including while a timed-out query is being drained. Keep successful + // results (especially commits), but don't report shutdown as a failure. + if result.is_err() && cancel.is_cancelled() { + Err(MaintenanceError::Cancelled) + } else { + result } } @@ -164,3 +172,56 @@ pub(super) async fn cancel_backend(pool: &PgPool, backend_pid: i32) { } } } + +#[cfg(test)] +mod tests { + use sqlx::postgres::PgPoolOptions; + use tokio_util::sync::CancellationToken; + + use super::{MaintenanceError, cancellable_within}; + + #[tokio::test] + async fn completed_query_error_during_cancellation_is_cancelled() { + let pool = PgPoolOptions::new() + .connect_lazy("postgres://localhost/unused") + .unwrap(); + let cancel = CancellationToken::new(); + cancel.cancel(); + + let result = cancellable_within(&pool, 0, &cancel, None, async { + Err::<(), _>(sqlx::Error::PoolClosed) + }) + .await; + + assert!(matches!(result, Err(MaintenanceError::Cancelled))); + } + + #[tokio::test] + async fn completed_query_error_without_cancellation_is_reported() { + let pool = PgPoolOptions::new() + .connect_lazy("postgres://localhost/unused") + .unwrap(); + let cancel = CancellationToken::new(); + + let result = cancellable_within(&pool, 0, &cancel, None, async { + Err::<(), _>(sqlx::Error::PoolClosed) + }) + .await; + + assert!(matches!(result, Err(MaintenanceError::Failed(_)))); + } + + #[tokio::test] + async fn completed_query_success_during_cancellation_is_preserved() { + let pool = PgPoolOptions::new() + .connect_lazy("postgres://localhost/unused") + .unwrap(); + let cancel = CancellationToken::new(); + cancel.cancel(); + + let result = + cancellable_within(&pool, 0, &cancel, None, async { Ok::<_, sqlx::Error>(42) }).await; + + assert_eq!(result.unwrap(), 42); + } +} From 448d6e15de27fac5f55a09826871381892a74474 Mon Sep 17 00:00:00 2001 From: Brandur Date: Thu, 5 Nov 2026 13:40:40 -0600 Subject: [PATCH 2/2] [go, rust] Guard reindexer shutdown logs and share startup checks --- internal/maintenance/job_cleaner.go | 5 +-- internal/maintenance/job_rescuer.go | 5 +-- internal/maintenance/job_scheduler.go | 5 +-- internal/maintenance/periodic_job_enqueuer.go | 5 +-- internal/maintenance/queue_cleaner.go | 5 +-- internal/maintenance/reindexer.go | 5 +-- internal/maintenance/shutdown_test.go | 14 +++++-- .../sqlite_notification_cleaner.go | 5 +-- .../river_shared_maintenance.go | 13 +++--- .../river_shared_maintenance_test.go | 40 +++++++++++++++++++ rust/riverqueue/src/maintenance/reindexer.rs | 6 +++ 11 files changed, 71 insertions(+), 37 deletions(-) create mode 100644 rivershared/riversharedmaintenance/river_shared_maintenance_test.go diff --git a/internal/maintenance/job_cleaner.go b/internal/maintenance/job_cleaner.go index ed7e29848..68863a232 100644 --- a/internal/maintenance/job_cleaner.go +++ b/internal/maintenance/job_cleaner.go @@ -145,12 +145,9 @@ func (s *JobCleaner) Start(ctx context.Context) error { //nolint:dupl return nil } - s.StaggerStart(ctx) - select { - case <-ctx.Done(): + if !s.StaggerStart(ctx) { stopped() return nil - default: } go func() { diff --git a/internal/maintenance/job_rescuer.go b/internal/maintenance/job_rescuer.go index a00269451..ebb1d47ef 100644 --- a/internal/maintenance/job_rescuer.go +++ b/internal/maintenance/job_rescuer.go @@ -142,12 +142,9 @@ func (s *JobRescuer) Start(ctx context.Context) error { return nil } - s.StaggerStart(ctx) - select { - case <-ctx.Done(): + if !s.StaggerStart(ctx) { stopped() return nil - default: } go func() { diff --git a/internal/maintenance/job_scheduler.go b/internal/maintenance/job_scheduler.go index d099ff067..09ce2d220 100644 --- a/internal/maintenance/job_scheduler.go +++ b/internal/maintenance/job_scheduler.go @@ -113,12 +113,9 @@ func (s *JobScheduler) Start(ctx context.Context) error { //nolint:dupl return nil } - s.StaggerStart(ctx) - select { - case <-ctx.Done(): + if !s.StaggerStart(ctx) { stopped() return nil - default: } go func() { diff --git a/internal/maintenance/periodic_job_enqueuer.go b/internal/maintenance/periodic_job_enqueuer.go index ea6aa1e2c..8292b074c 100644 --- a/internal/maintenance/periodic_job_enqueuer.go +++ b/internal/maintenance/periodic_job_enqueuer.go @@ -320,12 +320,9 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { return nil } - s.StaggerStart(ctx) - select { - case <-ctx.Done(): + if !s.StaggerStart(ctx) { stopped() return nil - default: } var ( diff --git a/internal/maintenance/queue_cleaner.go b/internal/maintenance/queue_cleaner.go index b81d06d53..3827fe678 100644 --- a/internal/maintenance/queue_cleaner.go +++ b/internal/maintenance/queue_cleaner.go @@ -106,12 +106,9 @@ func (s *QueueCleaner) Start(ctx context.Context) error { return nil } - s.StaggerStart(ctx) - select { - case <-ctx.Done(): + if !s.StaggerStart(ctx) { stopped() return nil - default: } go func() { diff --git a/internal/maintenance/reindexer.go b/internal/maintenance/reindexer.go index 1a6bcebdf..75b737855 100644 --- a/internal/maintenance/reindexer.go +++ b/internal/maintenance/reindexer.go @@ -113,12 +113,9 @@ func (s *Reindexer) Start(ctx context.Context) error { return nil } - s.StaggerStart(ctx) - select { - case <-ctx.Done(): + if !s.StaggerStart(ctx) { stopped() return nil - default: } go func() { diff --git a/internal/maintenance/shutdown_test.go b/internal/maintenance/shutdown_test.go index 4a51ca755..134451b65 100644 --- a/internal/maintenance/shutdown_test.go +++ b/internal/maintenance/shutdown_test.go @@ -30,8 +30,11 @@ type maintenanceShutdownExecutor struct { operationFunc func(context.Context, string) error } -func (e *maintenanceShutdownExecutor) IndexesExist(ctx context.Context, params *riverdriver.IndexesExistParams) (map[string]bool, error) { - return map[string]bool{"index": true}, e.operationFunc(ctx, "IndexesExist") +func (e *maintenanceShutdownExecutor) IndexesExist(ctx context.Context, _ *riverdriver.IndexesExistParams) (map[string]bool, error) { + if err := e.operationFunc(ctx, "IndexesExist"); err != nil { + return nil, err + } + return map[string]bool{"index": true}, nil } func (e *maintenanceShutdownExecutor) IndexReindex(ctx context.Context, _ *riverdriver.IndexReindexParams) error { @@ -47,7 +50,10 @@ func (e *maintenanceShutdownExecutor) JobDeleteBefore(ctx context.Context, _ *ri } func (e *maintenanceShutdownExecutor) JobGetStuck(ctx context.Context, _ *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) { - return []*rivertype.JobRow{{ID: 1, Kind: "unknown", Metadata: []byte(`{}`)}}, e.operationFunc(ctx, "JobGetStuck") + if err := e.operationFunc(ctx, "JobGetStuck"); err != nil { + return nil, err + } + return []*rivertype.JobRow{{ID: 1, Kind: "unknown", Metadata: []byte(`{}`)}}, nil } func (e *maintenanceShutdownExecutor) JobRescueMany(ctx context.Context, _ *riverdriver.JobRescueManyParams) (*struct{}, error) { @@ -189,6 +195,8 @@ func TestMaintenance_ShutdownErrorLogging(t *testing.T) { } // Exercise synchronous startup failures through the leader's real retry loop. +// The elector is intentionally nil: synctest.Wait does not advance the clock, +// so tests stop during the first backoff, before retries can exhaust and resign. func maintenanceStartupService(archetype *baseservice.Archetype, service startstop.Service) startstop.Service { leader := NewQueueMaintainerLeader(archetype, &QueueMaintainerLeaderConfig{ QueueMaintainer: NewQueueMaintainer(archetype, []startstop.Service{service}), diff --git a/internal/maintenance/sqlite_notification_cleaner.go b/internal/maintenance/sqlite_notification_cleaner.go index 232593dfa..16e80ece9 100644 --- a/internal/maintenance/sqlite_notification_cleaner.go +++ b/internal/maintenance/sqlite_notification_cleaner.go @@ -107,12 +107,9 @@ func (s *SQLiteNotificationCleaner) Start(ctx context.Context) error { //nolint: return nil } - s.StaggerStart(ctx) - select { - case <-ctx.Done(): + if !s.StaggerStart(ctx) { stopped() return nil - default: } go func() { diff --git a/rivershared/riversharedmaintenance/river_shared_maintenance.go b/rivershared/riversharedmaintenance/river_shared_maintenance.go index 824873001..dd9a0d611 100644 --- a/rivershared/riversharedmaintenance/river_shared_maintenance.go +++ b/rivershared/riversharedmaintenance/river_shared_maintenance.go @@ -124,13 +124,14 @@ type QueueMaintainerServiceBase struct { // StaggerStart is called when queue maintainer services start. It jitters by // sleeping for a short random period so services don't all perform their first -// run at exactly the same time. -func (s *QueueMaintainerServiceBase) StaggerStart(ctx context.Context) { - if s.staggerStartupDisabled { - return +// run at exactly the same time. Returns false if the context was cancelled +// before or during the sleep, in which case the caller should mark itself +// stopped and return instead of starting its run loop. +func (s *QueueMaintainerServiceBase) StaggerStart(ctx context.Context) bool { + if !s.staggerStartupDisabled { + serviceutil.CancellableSleep(ctx, randutil.DurationBetween(0*time.Second, 1*time.Second)) } - - serviceutil.CancellableSleep(ctx, randutil.DurationBetween(0*time.Second, 1*time.Second)) + return ctx.Err() == nil } // StaggerStartupDisable sets whether the short staggered sleep on start up diff --git a/rivershared/riversharedmaintenance/river_shared_maintenance_test.go b/rivershared/riversharedmaintenance/river_shared_maintenance_test.go new file mode 100644 index 000000000..73f7064f9 --- /dev/null +++ b/rivershared/riversharedmaintenance/river_shared_maintenance_test.go @@ -0,0 +1,40 @@ +package riversharedmaintenance + +import ( + "context" + "testing" + "testing/synctest" + + "github.com/stretchr/testify/require" +) + +func TestQueueMaintainerServiceBase_StaggerStart(t *testing.T) { + t.Parallel() + + for _, testCase := range []struct { + name string + cancelled bool + staggerDisabled bool + }{ + {name: "Cancelled", cancelled: true}, + {name: "CancelledWithStaggerDisabled", cancelled: true, staggerDisabled: true}, + {name: "StaggerDisabled", staggerDisabled: true}, + {name: "StaggerEnabled"}, + } { + t.Run(testCase.name, func(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + svc := &QueueMaintainerServiceBase{} + svc.StaggerStartupDisable(testCase.staggerDisabled) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if testCase.cancelled { + cancel() + } + + require.Equal(t, !testCase.cancelled, svc.StaggerStart(ctx)) + }) + }) + } +} diff --git a/rust/riverqueue/src/maintenance/reindexer.rs b/rust/riverqueue/src/maintenance/reindexer.rs index 911325661..db56be40e 100644 --- a/rust/riverqueue/src/maintenance/reindexer.rs +++ b/rust/riverqueue/src/maintenance/reindexer.rs @@ -97,6 +97,9 @@ pub(super) async fn run(context: std::sync::Arc, pool: sqlx::PgP Ok(false) => {} Err(MaintenanceError::Cancelled) => return, Err(reindex_error) => { + if context.cancel.is_cancelled() { + return; + } error!(error = %crate::error::Chain(&reindex_error), index_name, "River reindexer failed"); } } @@ -104,6 +107,9 @@ pub(super) async fn run(context: std::sync::Arc, pool: sqlx::PgP } Err(MaintenanceError::Cancelled) => return, Err(list_error) => { + if context.cancel.is_cancelled() { + return; + } error!(error = %crate::error::Chain(&list_error), "River reindexer could not list indexes"); } }