diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fd23289..2db6de46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed SQLite job list pagination skipping or repeating jobs by formatting cursor timestamps consistently with stored timestamps. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Improved PostgreSQL job listing performance when filtering by one finalized state (`completed`, `cancelled`, or `discarded`) and sorting by finalized time, including in River UI. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Fixed `JobRescuer` overwriting jobs that complete, leave the running state, or are claimed again by another worker after being fetched for rescue, preserving their state, errors, metadata, and timestamps across PostgreSQL and SQLite drivers. Fixes [#1302](https://github.com/riverqueue/river/issues/1302). [PR #1373](https://github.com/riverqueue/river/pull/1373). +- Attempt errors that are valid JSON but don't have the shape River writes (for example an `at` timestamp in another format, an `attempt` stored as a string, or an `error` or `trace` that isn't a string) are now decoded on a best effort basis instead of failing. Previously a single such element made its job unreadable, and if that happened while fetching jobs, every job locked in the same fetch was left `running` indefinitely. [PR #1380](https://github.com/riverqueue/river/pull/1380). +- A fetched job whose row can't be decoded (for example a SQLite job whose `tags` were changed to something other than an array of strings) no longer leaves every job locked in the same fetch stuck `running`. The other jobs are worked normally, while the undecodable job's attempt fails with an error describing the decode failure, and it's retried or discarded like any other failed job. Its stored values are left in place, and neither the rescuer nor the SQLite scheduler fails any longer on jobs that can't be decoded, so such a job's retry doesn't stop other jobs from being rescued or scheduled. [PR #1380](https://github.com/riverqueue/river/pull/1380). ## [0.47.0] - 2026-09-01 diff --git a/client_pilot_test.go b/client_pilot_test.go index 55e81c1a..fc9bc160 100644 --- a/client_pilot_test.go +++ b/client_pilot_test.go @@ -69,7 +69,7 @@ func (p *pilotSpy) JobCleanerQueuesExcluded() []string { return p.StandardPilot.JobCleanerQueuesExcluded() } -func (p *pilotSpy) JobGetAvailable(ctx context.Context, exec riverdriver.Executor, state riverpilot.ProducerState, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { +func (p *pilotSpy) JobGetAvailable(ctx context.Context, exec riverdriver.Executor, state riverpilot.ProducerState, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { p.testSignals.JobGetAvailable.Signal(struct{}{}) return p.StandardPilot.JobGetAvailable(ctx, exec, state, params) } diff --git a/internal/jobexecutor/job_executor.go b/internal/jobexecutor/job_executor.go index adaa6265..7f5bca91 100644 --- a/internal/jobexecutor/job_executor.go +++ b/internal/jobexecutor/job_executor.go @@ -115,7 +115,15 @@ type JobExecutor struct { PluginLookupByJob *pluginlookup.JobPluginLookup PluginLookupGlobal *pluginlookup.PluginLookup JobRow *rivertype.JobRow - ProducerCallbacks struct { + + // JobRowDecodeErr is set for a locked job whose row couldn't be fully + // decoded, in which case JobRow contains only the fields that could be. + // The job isn't worked. Instead, its attempt fails with an error + // describing the decode failure, and it's retried or discarded like any + // other failed attempt. + JobRowDecodeErr error + + ProducerCallbacks struct { JobDone func(jobRow *rivertype.JobRow) Stuck func(ctx context.Context, jobRow *rivertype.JobRow) Unstuck func() @@ -211,6 +219,15 @@ func (e *JobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { e.stats.RunDuration = e.Time.Now().Sub(e.start) }() + if e.JobRowDecodeErr != nil { + e.Logger.ErrorContext(ctx, e.Name+": Job row couldn't be decoded; failing attempt without working it", + slog.String("error", e.JobRowDecodeErr.Error()), + slog.Int64("job_id", e.JobRow.ID), + slog.String("kind", e.JobRow.Kind), + ) + return &jobExecutorResult{Err: fmt.Errorf("job row couldn't be decoded: %w", e.JobRowDecodeErr), MetadataUpdates: metadataUpdates} + } + if e.WorkUnit == nil { e.Logger.ErrorContext(ctx, e.Name+": Unhandled job kind", slog.String("kind", e.JobRow.Kind), diff --git a/internal/jobexecutor/job_executor_test.go b/internal/jobexecutor/job_executor_test.go index 34b4bb0f..f00ca772 100644 --- a/internal/jobexecutor/job_executor_test.go +++ b/internal/jobexecutor/job_executor_test.go @@ -159,13 +159,14 @@ func TestJobExecutor_Execute(t *testing.T) { require.NoError(t, err) // Fetch the job to make sure it's marked as running: - jobs, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + res, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ MaxToLock: 1, Now: new(now), Queue: rivercommon.QueueDefault, }) require.NoError(t, err) + jobs := res.Jobs require.Len(t, jobs, 1) require.Equal(t, results[0].Job.ID, jobs[0].ID) job := jobs[0] @@ -562,6 +563,71 @@ func TestJobExecutor_Execute(t *testing.T) { require.False(t, workCalled) }) + // A job whose row couldn't be decoded isn't worked. Its attempt fails with + // the decode error and goes through normal error handling. + t.Run("JobRowDecodeErrFailsAttemptWithoutWorking", func(t *testing.T) { + t.Parallel() + + executor, bundle := setup(t) + executor.ClientRetryPolicy = &retrypolicytest.RetryPolicyCustom{} + + var workCalled bool + executor.WorkUnit = &customizableWorkUnit{ + work: func() error { + workCalled = true + return nil + }, + } + + decodeErr := errors.New("error unmarshaling `tags`") + executor.JobRowDecodeErr = decodeErr + + bundle.errorHandler.HandleErrorFunc = func(ctx context.Context, job *rivertype.JobRow, err error) *ErrorHandlerResult { + require.ErrorIs(t, err, decodeErr) + return nil + } + + expectedRetryAt := executor.ClientRetryPolicy.NextRetry(bundle.jobRow) + + executor.Execute(ctx) + jobUpdates := riversharedtest.WaitOrTimeout(t, bundle.updateCh) + require.Len(t, jobUpdates, 1) + require.Equal(t, riverdriver.JobSetStateReasonFailed, jobUpdates[0].Reason) + + job, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ + ID: bundle.jobRow.ID, + Schema: "", + }) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateRetryable, job.State) + require.Len(t, job.Errors, 1) + require.Equal(t, bundle.jobRow.Attempt, job.Errors[0].Attempt) + require.Equal(t, "job row couldn't be decoded: error unmarshaling `tags`", job.Errors[0].Error) + require.WithinDuration(t, expectedRetryAt, job.ScheduledAt, time.Microsecond) + require.True(t, bundle.errorHandler.HandleErrorCalled) + require.False(t, workCalled) + }) + + t.Run("JobRowDecodeErrDiscardsJobAfterTooManyAttempts", func(t *testing.T) { + t.Parallel() + + executor, bundle := setup(t) + + bundle.jobRow.Attempt = bundle.jobRow.MaxAttempts + executor.JobRowDecodeErr = errors.New("error unmarshaling `tags`") + + executor.Execute(ctx) + riversharedtest.WaitOrTimeout(t, bundle.updateCh) + + job, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ + ID: bundle.jobRow.ID, + Schema: "", + }) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateDiscarded, job.State) + require.Equal(t, "job row couldn't be decoded: error unmarshaling `tags`", job.Errors[0].Error) + }) + t.Run("InvalidNextRetryAt", func(t *testing.T) { t.Parallel() @@ -676,8 +742,8 @@ func TestJobExecutor_Execute(t *testing.T) { Queue: rivercommon.QueueDefault, }) require.NoError(t, err) - require.Len(t, locked, 3) - return locked + require.Len(t, locked.Jobs, 3) + return locked.Jobs } t.Run("AllJobsShareSameNormalError", func(t *testing.T) { diff --git a/producer.go b/producer.go index da11246e..073bb1b9 100644 --- a/producer.go +++ b/producer.go @@ -658,10 +658,10 @@ func (p *producer) innerFetchLoop(workCtx context.Context, fetchResultCh chan pr case result := <-fetchResultCh: if result.err != nil { p.Logger.ErrorContext(workCtx, p.Name+": Error fetching jobs", slog.String("err", result.err.Error()), slog.String("queue", p.config.Queue)) - } else if len(result.jobs) > 0 { - p.startNewExecutors(workCtx, result.jobs) + } else if numLocked := len(result.jobs) + len(result.undecodableJobs); numLocked > 0 { + p.startNewExecutors(workCtx, result.jobs, result.undecodableJobs) - if len(result.jobs) == limit { + if numLocked == limit { // Fetch returned the maximum number of jobs that were requested, // implying there may be more in the queue. Trigger another fetch when // slots are available. @@ -833,7 +833,7 @@ func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultC startedAt = time.Now() } - jobs, err := p.pilot.JobGetAvailable(ctx, p.exec, p.state, &riverdriver.JobGetAvailableParams{ + res, err := p.pilot.JobGetAvailable(ctx, p.exec, p.state, &riverdriver.JobGetAvailableParams{ ClientID: p.config.ClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: count, @@ -846,6 +846,9 @@ func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultC fetchResultCh <- producerFetchResult{err: err} return } + if res == nil { // tolerate a pilot returning a nil result when it locks no jobs + res = &riverdriver.JobGetAvailableResult{} + } if len(p.metricEmitHooks) > 0 { p.emitMetric(ctx, &rivertype.HookMetricEmitParams{ @@ -856,13 +859,13 @@ func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultC }) p.emitMetric(ctx, &rivertype.HookMetricEmitParams{ Metric: &rivertype.JobGetAvailableCountMetric{ - Count: len(jobs), + Count: len(res.Jobs) + len(res.UndecodableJobs), Queue: p.config.Queue, }, }) } - fetchResultCh <- producerFetchResult{jobs: jobs} + fetchResultCh <- producerFetchResult{jobs: res.Jobs, undecodableJobs: res.UndecodableJobs} } func (p *producer) emitMetric(ctx context.Context, params *rivertype.HookMetricEmitParams) { @@ -907,15 +910,19 @@ func (p *producer) heartbeatLogLoop(ctx context.Context, wg *sync.WaitGroup) { } } -func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.JobRow) { +// startNewExecutors starts an executor for each locked job. A job whose row +// couldn't be decoded isn't worked, but it still gets an executor that fails +// its attempt with the decode error, so that it's retried or discarded +// through the normal error handling path instead of being left running. +func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.JobRow, undecodableJobs []*riverdriver.UndecodableJob) { defaultClientRetryPolicy := retrypolicy.NewDefault(p.Time) - for _, job := range jobs { - workInfo, ok := p.workers.workersMap[job.Kind] - + startExecutor := func(job *rivertype.JobRow, decodeErr error) { var workUnit workunit.WorkUnit - if ok { - workUnit = workInfo.workUnitFactory.MakeUnit(job) + if decodeErr == nil { + if workInfo, ok := p.workers.workersMap[job.Kind]; ok { + workUnit = workInfo.workUnitFactory.MakeUnit(job) + } } // jobCancel will always be called by the executor to prevent leaks. @@ -932,6 +939,7 @@ func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype. PluginLookupByJob: p.config.PluginLookupByJob, PluginLookupGlobal: p.config.PluginLookupGlobal, JobRow: job, + JobRowDecodeErr: decodeErr, ProducerCallbacks: struct { JobDone func(jobRow *rivertype.JobRow) Stuck func(ctx context.Context, jobRow *rivertype.JobRow) @@ -950,7 +958,14 @@ func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype. go executor.Execute(jobCtx) } - p.Logger.DebugContext(workCtx, p.Name+": Distributed batch of jobs to executors", "num_jobs", len(jobs)) + for _, job := range jobs { + startExecutor(job, nil) + } + for _, undecodableJob := range undecodableJobs { + startExecutor(undecodableJob.Job, undecodableJob.DecodeErr) + } + + p.Logger.DebugContext(workCtx, p.Name+": Distributed batch of jobs to executors", "num_jobs", len(jobs)+len(undecodableJobs)) p.testSignals.StartedExecutors.Signal(struct{}{}) } @@ -1118,8 +1133,9 @@ func (p *producer) reportQueueStatusOnce(ctx context.Context) { } type producerFetchResult struct { - jobs []*rivertype.JobRow - err error + jobs []*rivertype.JobRow + err error + undecodableJobs []*riverdriver.UndecodableJob } type errorHandlerAdapter struct { diff --git a/producer_test.go b/producer_test.go index a7041433..a8e34dd6 100644 --- a/producer_test.go +++ b/producer_test.go @@ -3,6 +3,7 @@ package river import ( "context" "encoding/json" + "errors" "fmt" "slices" "sync/atomic" @@ -45,7 +46,7 @@ func (p *beforeJobGetAvailablePilot) JobGetAvailable( exec riverdriver.Executor, state riverpilot.ProducerState, params *riverdriver.JobGetAvailableParams, -) ([]*rivertype.JobRow, error) { +) (*riverdriver.JobGetAvailableResult, error) { if p.beforeJobGetAvailableFunc != nil { p.beforeJobGetAvailableFunc(params) } @@ -53,6 +54,39 @@ func (p *beforeJobGetAvailablePilot) JobGetAvailable( return p.Pilot.JobGetAvailable(ctx, exec, state, params) } +// undecodableKindPilot reports locked jobs of one kind as undecodable. Postgres' +// column types don't allow a job row that can't be decoded, so this simulates +// one. +type undecodableKindPilot struct { + riverpilot.Pilot + + kind string +} + +func (p *undecodableKindPilot) JobGetAvailable( + ctx context.Context, + exec riverdriver.Executor, + state riverpilot.ProducerState, + params *riverdriver.JobGetAvailableParams, +) (*riverdriver.JobGetAvailableResult, error) { + res, err := p.Pilot.JobGetAvailable(ctx, exec, state, params) + if err != nil { + return nil, err + } + + jobs := make([]*rivertype.JobRow, 0, len(res.Jobs)) + for _, job := range res.Jobs { + if job.Kind == p.kind { + res.UndecodableJobs = append(res.UndecodableJobs, &riverdriver.UndecodableJob{DecodeErr: errors.New("fake decode error"), Job: job}) + continue + } + jobs = append(jobs, job) + } + res.Jobs = jobs + + return res, nil +} + func TestProducer_MetricEmitHook(t *testing.T) { t.Parallel() @@ -458,6 +492,50 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin } }) + // A locked job whose row can't be decoded isn't worked. Its attempt fails + // with the decode error, and the other jobs locked with it are worked + // normally. + t.Run("UndecodableJob", func(t *testing.T) { + t.Parallel() + + producer, bundle := setup(t) + + type JobArgs struct { + testutil.JobArgsReflectKind[JobArgs] + } + + AddWorker(bundle.workers, &noOpWorker{}) + AddWorker(bundle.workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error { + t.Error("undecodable job shouldn't be worked") // not FailNow because this runs outside the test goroutine + return nil + })) + + producer.pilot = &undecodableKindPilot{ + Pilot: producer.pilot, + kind: (&JobArgs{}).Kind(), + } + + mustInsert(ctx, t, producer, bundle, &noOpArgs{}) + mustInsert(ctx, t, producer, bundle, &JobArgs{}) + mustInsert(ctx, t, producer, bundle, &noOpArgs{}) + + startProducer(t, ctx, ctx, producer) + + updates := riversharedtest.WaitOrTimeoutN(t, bundle.jobUpdates, 3) + + for _, update := range updates { + if update.Job.Kind == (&JobArgs{}).Kind() { + require.Equal(t, rivertype.JobStateRetryable, update.Job.State) + require.Len(t, update.Job.Errors, 1) + require.Equal(t, 1, update.Job.Errors[0].Attempt) + require.Equal(t, "job row couldn't be decoded: fake decode error", update.Job.Errors[0].Error) + continue + } + + require.Equal(t, rivertype.JobStateCompleted, update.Job.State) + } + }) + t.Run("CancelledWorkContextCancelsJob", func(t *testing.T) { t.Parallel() diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index d4b87f31..60ba3c50 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -231,11 +231,24 @@ type Executor interface { JobDelete(ctx context.Context, params *JobDeleteParams) (*rivertype.JobRow, error) JobDeleteBefore(ctx context.Context, params *JobDeleteBeforeParams) (int, error) JobDeleteMany(ctx context.Context, params *JobDeleteManyParams) ([]*rivertype.JobRow, error) - JobGetAvailable(ctx context.Context, params *JobGetAvailableParams) ([]*rivertype.JobRow, error) + + // JobGetAvailable locks available jobs for work, moving them to `running`. + // A locked job whose row can't be fully decoded doesn't fail the call. + // It's instead returned in the result's UndecodableJobs so that the caller + // can fail the attempt, since it's been moved to `running` along with the + // others. + JobGetAvailable(ctx context.Context, params *JobGetAvailableParams) (*JobGetAvailableResult, error) + JobGetByID(ctx context.Context, params *JobGetByIDParams) (*rivertype.JobRow, error) JobGetByIDMany(ctx context.Context, params *JobGetByIDManyParams) ([]*rivertype.JobRow, error) JobGetByKindMany(ctx context.Context, params *JobGetByKindManyParams) ([]*rivertype.JobRow, error) + + // JobGetStuck gets jobs that have been running since before a horizon. A + // job row that can't be fully decoded is returned with the fields that + // couldn't be decoded left empty so that one bad row can't prevent stuck + // jobs from being rescued. JobGetStuck(ctx context.Context, params *JobGetStuckParams) ([]*rivertype.JobRow, error) + JobInsertFastMany(ctx context.Context, params *JobInsertFastManyParams) ([]*JobInsertFastResult, error) JobInsertFastManyNoReturning(ctx context.Context, params *JobInsertFastManyParams) (int, error) JobInsertFull(ctx context.Context, params *JobInsertFullParams) (*rivertype.JobRow, error) @@ -245,7 +258,13 @@ type Executor interface { JobRescueMany(ctx context.Context, params *JobRescueManyParams) (*struct{}, error) JobRetry(ctx context.Context, params *JobRetryParams) (*rivertype.JobRow, error) JobSchedule(ctx context.Context, params *JobScheduleParams) ([]*JobScheduleResult, error) + + // JobSetStateIfRunningMany sets the state of running jobs, returning the + // resulting rows. A job row that can't be fully decoded is returned with + // the fields that couldn't be decoded left empty so that the state of an + // undecodable job can be set without failing the other jobs set with it. JobSetStateIfRunningMany(ctx context.Context, params *JobSetStateIfRunningManyParams) ([]*rivertype.JobRow, error) + JobUpdate(ctx context.Context, params *JobUpdateParams) (*rivertype.JobRow, error) JobUpdateFull(ctx context.Context, params *JobUpdateFullParams) (*rivertype.JobRow, error) LeaderAttemptElect(ctx context.Context, params *LeaderElectParams) (*Leader, error) @@ -432,6 +451,17 @@ type JobGetAvailableParams struct { Schema string } +// JobGetAvailableResult is the result of JobGetAvailable. +type JobGetAvailableResult struct { + // Jobs are the locked jobs that were decoded successfully. + Jobs []*rivertype.JobRow + + // UndecodableJobs are locked jobs whose rows couldn't be fully decoded. + // They've been moved to `running` like Jobs, so the caller should fail + // their attempt rather than leave them for the rescuer. + UndecodableJobs []*UndecodableJob +} + type JobGetByIDParams struct { ID int64 Schema string @@ -948,6 +978,18 @@ type TableTruncateParams struct { Table []string } +// UndecodableJob is a job that was locked by JobGetAvailable, but whose row +// couldn't be fully decoded, like when one of its JSON columns has been +// changed to a shape that doesn't match its JobRow field. +type UndecodableJob struct { + // DecodeErr describes why the job row couldn't be decoded. + DecodeErr error + + // Job is the job row with every field that could be decoded. Fields that + // couldn't be decoded are left empty. + Job *rivertype.JobRow +} + // MigrationLineMainTruncateTables is a shared helper that produces tables to // truncate for the main migration line. It's reused across all drivers. // diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go index 8677d0fd..79af28cb 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/riverdriver/riverdatabasesql/river_database_sql_driver.go @@ -351,7 +351,7 @@ func (e *Executor) JobDeleteMany(ctx context.Context, params *riverdriver.JobDel return sliceutil.MapError(jobs, jobRowFromInternal) } -func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { +func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { jobs, err := dbsqlc.New().JobGetAvailable(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetAvailableParams{ AttemptedBy: params.ClientID, MaxAttemptedBy: int32(min(params.MaxAttemptedBy, math.MaxInt32)), //nolint:gosec @@ -362,7 +362,7 @@ func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobG if err != nil { return nil, interpretError(err) } - return sliceutil.MapError(jobs, jobRowFromInternal) + return jobGetAvailableResultFromInternal(jobs), nil } func (e *Executor) JobGetByID(ctx context.Context, params *riverdriver.JobGetByIDParams) (*rivertype.JobRow, error) { @@ -398,7 +398,7 @@ func (e *Executor) JobGetStuck(ctx context.Context, params *riverdriver.JobGetSt if err != nil { return nil, interpretError(err) } - return sliceutil.MapError(jobs, jobRowFromInternal) + return jobRowsFromInternalPartial(jobs), nil } func (e *Executor) JobInsertFastMany(ctx context.Context, params *riverdriver.JobInsertFastManyParams) ([]*riverdriver.JobInsertFastResult, error) { @@ -724,7 +724,7 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr if err != nil { return nil, interpretError(err) } - return sliceutil.MapError(jobs, jobRowFromInternal) + return jobRowsFromInternalPartial(jobs), nil } func (e *Executor) JobUpdate(ctx context.Context, params *riverdriver.JobUpdateParams) (*rivertype.JobRow, error) { @@ -1229,17 +1229,50 @@ func intToInt32(value int) (int32, error) { return int32(value), nil } +// jobGetAvailableResultFromInternal decodes the job rows locked by +// JobGetAvailable, separating out any that can't be decoded rather than +// failing all of them, because they've all been moved to `running`. +func jobGetAvailableResultFromInternal(jobs []*dbsqlc.RiverJob) *riverdriver.JobGetAvailableResult { + res := &riverdriver.JobGetAvailableResult{Jobs: make([]*rivertype.JobRow, 0, len(jobs))} + for _, internal := range jobs { + job, err := jobRowFromInternalPartial(internal) + if err != nil { + res.UndecodableJobs = append(res.UndecodableJobs, &riverdriver.UndecodableJob{DecodeErr: err, Job: job}) + continue + } + res.Jobs = append(res.Jobs, job) + } + return res +} + +// jobRowFromInternal decodes a job row, returning an error if any of its +// fields can't be decoded. func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { + job, err := jobRowFromInternalPartial(internal) + if err != nil { + return nil, err + } + return job, nil +} + +// jobRowFromInternalPartial decodes a job row. A row is always returned, even +// along with an error, in which case the fields that couldn't be decoded are +// left empty. +func jobRowFromInternalPartial(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { var attemptedAt *time.Time if internal.AttemptedAt != nil { t := internal.AttemptedAt.UTC() attemptedAt = &t } - errors := make([]rivertype.AttemptError, len(internal.Errors)) + var decodeErrs []error + + attemptErrors := make([]rivertype.AttemptError, len(internal.Errors)) for i, rawError := range internal.Errors { - if err := json.Unmarshal([]byte(rawError), &errors[i]); err != nil { - return nil, err + if err := json.Unmarshal([]byte(rawError), &attemptErrors[i]); err != nil { + decodeErrs = append(decodeErrs, fmt.Errorf("error unmarshaling `errors`: %w", err)) + attemptErrors = nil + break } } @@ -1251,7 +1284,7 @@ func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { uniqueStates, err := intToByte(bitIntegerToBits(ptrutil.ValOrDefault(internal.UniqueStates, 0), 8)) if err != nil { - return nil, err + decodeErrs = append(decodeErrs, fmt.Errorf("error decoding `unique_states`: %w", err)) } return &rivertype.JobRow{ @@ -1261,7 +1294,7 @@ func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { AttemptedBy: internal.AttemptedBy, CreatedAt: internal.CreatedAt.UTC(), EncodedArgs: []byte(internal.Args), - Errors: errors, + Errors: attemptErrors, FinalizedAt: finalizedAt, Kind: internal.Kind, MaxAttempts: max(int(internal.MaxAttempts), 0), @@ -1273,7 +1306,17 @@ func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { Tags: internal.Tags, UniqueKey: internal.UniqueKey, UniqueStates: uniquestates.UniqueBitmaskToStates(uniqueStates), - }, nil + }, errors.Join(decodeErrs...) +} + +// jobRowsFromInternalPartial decodes job rows with jobRowFromInternalPartial, +// ignoring decode errors so that one bad row doesn't prevent returning the +// others. +func jobRowsFromInternalPartial(jobs []*dbsqlc.RiverJob) []*rivertype.JobRow { + return sliceutil.Map(jobs, func(internal *dbsqlc.RiverJob) *rivertype.JobRow { + job, _ := jobRowFromInternalPartial(internal) + return job + }) } func leaderFromInternal(internal *dbsqlc.RiverLeader) *riverdriver.Leader { diff --git a/riverdriver/riverdrivertest/benchmark.go b/riverdriver/riverdrivertest/benchmark.go index d0118d3e..dde1e086 100644 --- a/riverdriver/riverdrivertest/benchmark.go +++ b/riverdriver/riverdrivertest/benchmark.go @@ -259,10 +259,11 @@ func Benchmark[TTx any](ctx context.Context, b *testing.B, b.ResetTimer() for range b.N { - jobs, err := exec.JobGetAvailable(ctx, getAvailableParams) + res, err := exec.JobGetAvailable(ctx, getAvailableParams) if err != nil { b.Fatalf("failed to fetch benchmark job: %v", err) } + jobs := res.Jobs if len(jobs) != 1 { b.Fatalf("expected exactly one fetched job, got %d", len(jobs)) } diff --git a/riverdriver/riverdrivertest/driver_client_test.go b/riverdriver/riverdrivertest/driver_client_test.go index f1d720eb..baaa0f34 100644 --- a/riverdriver/riverdrivertest/driver_client_test.go +++ b/riverdriver/riverdrivertest/driver_client_test.go @@ -265,6 +265,14 @@ func newTestConfig(t *testing.T, schema string) *river.Config { } } +// retryPolicyAnHourLater schedules every retry an hour out so that a failed job +// stays `retryable` for the rest of a test. +type retryPolicyAnHourLater struct{} + +func (*retryPolicyAnHourLater) NextRetry(job *rivertype.JobRow) time.Time { + return time.Now().Add(time.Hour) +} + // Try to keep this helper close to the one found in the top-level package so we // can copy/paste between them reasonably easily. func startClient[TTx any](ctx context.Context, t *testing.T, client *river.Client[TTx]) { @@ -1253,6 +1261,32 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T, require.Equal(t, job.ID, listRes.Jobs[0].ID) }) + // Attempt errors in an unexpected shape are decoded leniently, so the job + // is still worked. + t.Run("JobWithUnexpectedAttemptErrorsWorked", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + job := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{ + Errors: [][]byte{ + []byte(`{"at":"2024-01-02 03:04:05+00","attempt":"1","error":{"message":"boom"},"trace":["frame"]}`), + }, + Kind: new(noOpArgs{}.Kind()), + Schema: bundle.schema, + }) + + subscribeChan := subscribe(t, client) + startClient(ctx, t, client) + + event := riversharedtest.WaitOrTimeout(t, subscribeChan) + require.Equal(t, river.EventKindJobCompleted, event.Kind) + require.Equal(t, job.ID, event.Job.ID) + require.Len(t, event.Job.Errors, 1) + require.Equal(t, 1, event.Job.Errors[0].Attempt) + require.JSONEq(t, `{"message":"boom"}`, event.Job.Errors[0].Error) + }) + t.Run("QueueGet", func(t *testing.T) { t.Parallel() @@ -1398,4 +1432,63 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T, require.NoError(t, err) require.JSONEq(t, `{}`, string(fetchedQueue.Metadata)) }) + + // A locked job whose row can't be decoded has its attempt failed (and is + // retried or discarded like any other failed job) without preventing the + // jobs locked alongside it from being worked. + t.Run("UndecodableJobFailedWithoutBlockingOthers", func(t *testing.T) { + t.Parallel() + + config, bundle := setupConfig(t) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("only SQLite's JSON columns can hold values that don't decode") + } + config.RetryPolicy = &retryPolicyAnHourLater{} + + client, err := river.NewClient(bundle.driver, config) + require.NoError(t, err) + + var ( + goodJob1 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Kind: new(noOpArgs{}.Kind()), Schema: bundle.schema}) + undecodableJob1 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Kind: new(noOpArgs{}.Kind()), Schema: bundle.schema, Tags: []string{"tag"}}) + undecodableJob2 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Kind: new(noOpArgs{}.Kind()), MaxAttempts: new(1), Schema: bundle.schema}) + goodJob2 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Kind: new(noOpArgs{}.Kind()), Schema: bundle.schema}) + undecodableValue = `{"not":"an array"}` + ) + sqliteSetJobJSONColumn(ctx, t, bundle.exec, undecodableJob1.ID, "tags", undecodableValue) + sqliteSetJobJSONColumn(ctx, t, bundle.exec, undecodableJob2.ID, "tags", undecodableValue) + + subscribeChan := subscribe(t, client) + startClient(ctx, t, client) + + eventsByJobID := make(map[int64]*river.Event) + for range 4 { + event := riversharedtest.WaitOrTimeout(t, subscribeChan) + eventsByJobID[event.Job.ID] = event + } + + require.Equal(t, river.EventKindJobCompleted, eventsByJobID[goodJob1.ID].Kind) + require.Equal(t, river.EventKindJobCompleted, eventsByJobID[goodJob2.ID].Kind) + + // Events for undecodable jobs carry the fields that could be decoded. + require.Equal(t, river.EventKindJobFailed, eventsByJobID[undecodableJob1.ID].Kind) + require.Equal(t, rivertype.JobStateRetryable, eventsByJobID[undecodableJob1.ID].Job.State) + require.Nil(t, eventsByJobID[undecodableJob1.ID].Job.Tags) + require.Equal(t, river.EventKindJobFailed, eventsByJobID[undecodableJob2.ID].Kind) + require.Equal(t, rivertype.JobStateDiscarded, eventsByJobID[undecodableJob2.ID].Job.State) + + // The undecodable value is left as it was. + _, err = client.JobGet(ctx, undecodableJob1.ID) + require.ErrorContains(t, err, "error unmarshaling `tags`") + + // Once repaired, the job shows the failed attempt. + sqliteSetJobJSONColumn(ctx, t, bundle.exec, undecodableJob1.ID, "tags", `["tag"]`) + job, err := client.JobGet(ctx, undecodableJob1.ID) + require.NoError(t, err) + require.Equal(t, 1, job.Attempt) + require.Equal(t, rivertype.JobStateRetryable, job.State) + require.Len(t, job.Errors, 1) + require.Equal(t, 1, job.Errors[0].Attempt) + require.Contains(t, job.Errors[0].Error, "job row couldn't be decoded: error unmarshaling `tags`") + }) } diff --git a/riverdriver/riverdrivertest/job_read.go b/riverdriver/riverdrivertest/job_read.go index f8fe4418..7558fc3f 100644 --- a/riverdriver/riverdrivertest/job_read.go +++ b/riverdriver/riverdrivertest/job_read.go @@ -241,6 +241,16 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx maxToLock = 100 ) + // Gets available jobs, requiring that all of them were decoded. + getAvailable := func(t *testing.T, exec riverdriver.Executor, params *riverdriver.JobGetAvailableParams) []*rivertype.JobRow { + t.Helper() + + res, err := exec.JobGetAvailable(ctx, params) + require.NoError(t, err) + require.Empty(t, res.UndecodableJobs) + return res.Jobs + } + t.Run("Success", func(t *testing.T) { t.Parallel() @@ -248,13 +258,12 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{}) - jobRows, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: maxToLock, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Len(t, jobRows, 1) jobRow := jobRows[0] @@ -270,13 +279,12 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{}) // Two rows inserted but only one found because of the added limit. - jobRows, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: 1, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Len(t, jobRows, 1) }) @@ -290,13 +298,12 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx }) // Job is in a non-default queue so it's not found. - jobRows, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: maxToLock, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Empty(t, jobRows) }) @@ -312,14 +319,13 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx }) // Job is scheduled a while from now so it's not found. - jobRows, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: maxToLock, Now: &now, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Empty(t, jobRows) }) @@ -338,14 +344,13 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx ScheduledAt: new(now.Add(-1 * time.Microsecond)), }) - jobRows, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: maxToLock, Now: new(now), Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Len(t, jobRows, 1) require.Equal(t, job2.ID, jobRows[0].ID) }) @@ -362,13 +367,12 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx }) } - jobRows, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: 2, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Len(t, jobRows, 2, "expected to fetch exactly 2 jobs") // Because the jobs are ordered within the fetch query's CTE but *not* within @@ -382,14 +386,12 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx require.Equal(t, 2, jobRows[1].Priority, "expected second job to have priority 2") // Should fetch the one remaining job on the next attempt: - jobRows, err = exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows = getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: 1, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) - require.NoError(t, err) require.Len(t, jobRows, 1, "expected to fetch exactly 1 job") require.Equal(t, 3, jobRows[0].Priority, "expected final job to have priority 3") }) @@ -409,13 +411,12 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx }) // Job is in a non-default queue so it's not found. - jobRows, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: maxToLock, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Len(t, jobRows, 1) jobRow := jobRows[0] @@ -445,13 +446,12 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx }) // Job is in a non-default queue so it's not found. - jobRows, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, MaxAttemptedBy: maxAttemptedBy, MaxToLock: maxToLock, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Len(t, jobRows, 1) jobRow := jobRows[0] @@ -461,6 +461,90 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx ), jobRow.AttemptedBy) require.Len(t, jobRow.AttemptedBy, maxAttemptedBy) }) + + // Attempt errors written by something other than River may not have the + // shape River expects. They're decoded leniently so that the job (and + // every other job locked alongside it) can still be worked. + t.Run("ErrorsWithUnexpectedShapesDecoded", func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + job1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ + Errors: [][]byte{ + []byte(`{"at":"2024-01-02 03:04:05+00","attempt":"1","error":{"message":"boom"},"trace":["frame1","frame2"]}`), + []byte(`42`), + }, + }) + job2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{}) + + jobRows := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ + ClientID: testClientID, + MaxAttemptedBy: maxAttemptedBy, + MaxToLock: maxToLock, + Queue: rivercommon.QueueDefault, + }) + + // Result order isn't guaranteed by every driver. + sort.Slice(jobRows, func(i, j int) bool { return jobRows[i].ID < jobRows[j].ID }) + require.Equal(t, []int64{job1.ID, job2.ID}, + sliceutil.Map(jobRows, func(j *rivertype.JobRow) int64 { return j.ID })) + + require.Equal(t, []rivertype.AttemptError{ + { + At: time.Date(2024, 1, 2, 3, 4, 5, 0, time.UTC), + Attempt: 1, + Error: `{"message":"boom"}`, + Trace: `["frame1","frame2"]`, + }, + {Error: "42"}, + }, sliceutil.Map(jobRows[0].Errors, func(e rivertype.AttemptError) rivertype.AttemptError { + e.At = e.At.UTC() // normalize location of the fixed +00 offset + return e + })) + }) + + // A locked job whose row can't be decoded is returned separately so the + // caller can fail its attempt, and doesn't prevent returning the others. + t.Run("UndecodableJobsReturnedSeparately", func(t *testing.T) { + t.Parallel() + + exec, bundle := setup(ctx, t) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("only SQLite's JSON columns can hold values that don't decode") + } + + job1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{}) + job2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Tags: []string{"tag"}}) + job3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{}) + + sqliteSetJobJSONColumn(ctx, t, exec, job2.ID, "errors", `{"not":"an array"}`) + sqliteSetJobJSONColumn(ctx, t, exec, job2.ID, "tags", `{"not":"an array"}`) + + res, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + ClientID: testClientID, + MaxAttemptedBy: maxAttemptedBy, + MaxToLock: maxToLock, + Queue: rivercommon.QueueDefault, + }) + require.NoError(t, err) + require.Equal(t, []int64{job1.ID, job3.ID}, + sliceutil.Map(res.Jobs, func(j *rivertype.JobRow) int64 { return j.ID })) + + require.Len(t, res.UndecodableJobs, 1) + undecodableJob := res.UndecodableJobs[0] + require.ErrorContains(t, undecodableJob.DecodeErr, "error unmarshaling `errors`") + require.ErrorContains(t, undecodableJob.DecodeErr, "error unmarshaling `tags`") + + // Fields that could be decoded are set, while the others are empty. + require.Equal(t, job2.ID, undecodableJob.Job.ID) + require.Equal(t, 1, undecodableJob.Job.Attempt) + require.Equal(t, []string{testClientID}, undecodableJob.Job.AttemptedBy) + require.Equal(t, job2.Kind, undecodableJob.Job.Kind) + require.Equal(t, rivertype.JobStateRunning, undecodableJob.Job.State) + require.Nil(t, undecodableJob.Job.Errors) + require.Nil(t, undecodableJob.Job.Tags) + }) }) t.Run("JobGetByID", func(t *testing.T) { @@ -538,49 +622,83 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx t.Run("JobGetStuck", func(t *testing.T) { t.Parallel() - exec, _ := setup(ctx, t) + t.Run("Success", func(t *testing.T) { + t.Parallel() - var ( - horizon = time.Now().UTC() - beforeHorizon = horizon.Add(-1 * time.Minute) - afterHorizon = horizon.Add(1 * time.Minute) - ) + exec, _ := setup(ctx, t) + + var ( + horizon = time.Now().UTC() + beforeHorizon = horizon.Add(-1 * time.Minute) + afterHorizon = horizon.Add(1 * time.Minute) + ) - stuckJob1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: new(rivertype.JobStateRunning)}) - stuckJob2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: new(rivertype.JobStateRunning)}) + stuckJob1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: new(rivertype.JobStateRunning)}) + stuckJob2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: new(rivertype.JobStateRunning)}) - t.Logf("horizon = %s", horizon) - t.Logf("stuckJob1 = %s", stuckJob1.AttemptedAt) - t.Logf("stuckJob2 = %s", stuckJob2.AttemptedAt) + t.Logf("horizon = %s", horizon) + t.Logf("stuckJob1 = %s", stuckJob1.AttemptedAt) + t.Logf("stuckJob2 = %s", stuckJob2.AttemptedAt) - t.Logf("stuckJob1 full = %s", spew.Sdump(stuckJob1)) + t.Logf("stuckJob1 full = %s", spew.Sdump(stuckJob1)) - // Not returned on the first page because we put a maximum of two. - stuckJob3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: new(rivertype.JobStateRunning)}) + // Not returned on the first page because we put a maximum of two. + stuckJob3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: new(rivertype.JobStateRunning)}) - // Not stuck because not in running state. - _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: new(rivertype.JobStateAvailable)}) + // Not stuck because not in running state. + _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: new(rivertype.JobStateAvailable)}) - // Not stuck because after queried horizon. - _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &afterHorizon, State: new(rivertype.JobStateRunning)}) + // Not stuck because after queried horizon. + _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &afterHorizon, State: new(rivertype.JobStateRunning)}) - // Max two stuck - stuckJobs, err := exec.JobGetStuck(ctx, &riverdriver.JobGetStuckParams{ - Max: 2, - StuckHorizon: horizon, + // Max two stuck + stuckJobs, err := exec.JobGetStuck(ctx, &riverdriver.JobGetStuckParams{ + Max: 2, + StuckHorizon: horizon, + }) + require.NoError(t, err) + require.Equal(t, []int64{stuckJob1.ID, stuckJob2.ID}, + sliceutil.Map(stuckJobs, func(j *rivertype.JobRow) int64 { return j.ID })) + + stuckJobs, err = exec.JobGetStuck(ctx, &riverdriver.JobGetStuckParams{ + AfterID: stuckJob2.ID, + Max: 2, + StuckHorizon: horizon, + }) + require.NoError(t, err) + require.Equal(t, []int64{stuckJob3.ID}, + sliceutil.Map(stuckJobs, func(j *rivertype.JobRow) int64 { return j.ID })) }) - require.NoError(t, err) - require.Equal(t, []int64{stuckJob1.ID, stuckJob2.ID}, - sliceutil.Map(stuckJobs, func(j *rivertype.JobRow) int64 { return j.ID })) - stuckJobs, err = exec.JobGetStuck(ctx, &riverdriver.JobGetStuckParams{ - AfterID: stuckJob2.ID, - Max: 2, - StuckHorizon: horizon, + // A stuck job whose row can't be fully decoded is still returned so that + // it can be rescued. + t.Run("UndecodableJobReturned", func(t *testing.T) { + t.Parallel() + + exec, bundle := setup(ctx, t) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("only SQLite's JSON columns can hold values that don't decode") + } + + var ( + horizon = time.Now().UTC() + beforeHorizon = horizon.Add(-1 * time.Minute) + ) + + stuckJob1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: new(rivertype.JobStateRunning)}) + stuckJob2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &beforeHorizon, State: new(rivertype.JobStateRunning)}) + + sqliteSetJobJSONColumn(ctx, t, exec, stuckJob1.ID, "tags", `{"not":"an array"}`) + + stuckJobs, err := exec.JobGetStuck(ctx, &riverdriver.JobGetStuckParams{ + Max: 10, + StuckHorizon: horizon, + }) + require.NoError(t, err) + require.Equal(t, []int64{stuckJob1.ID, stuckJob2.ID}, + sliceutil.Map(stuckJobs, func(j *rivertype.JobRow) int64 { return j.ID })) + require.Nil(t, stuckJobs[0].Tags) }) - require.NoError(t, err) - require.Equal(t, []int64{stuckJob3.ID}, - sliceutil.Map(stuckJobs, func(j *rivertype.JobRow) int64 { return j.ID })) }) t.Run("JobKindList", func(t *testing.T) { diff --git a/riverdriver/riverdrivertest/job_update.go b/riverdriver/riverdrivertest/job_update.go index 2512ff31..70e393b8 100644 --- a/riverdriver/riverdrivertest/job_update.go +++ b/riverdriver/riverdrivertest/job_update.go @@ -16,6 +16,7 @@ import ( "github.com/riverqueue/river/riverdriver" "github.com/riverqueue/river/rivershared/testfactory" "github.com/riverqueue/river/rivershared/uniquestates" + "github.com/riverqueue/river/rivershared/util/sliceutil" "github.com/riverqueue/river/rivertype" ) @@ -359,7 +360,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT require.Len(t, releasedJobs, 1) require.Equal(t, rivertype.JobStateAvailable, releasedJobs[0].State) - claimedJobs, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + claimRes, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ ClientID: "new-worker", MaxAttemptedBy: 10, MaxToLock: 1, @@ -367,6 +368,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT Queue: job.Queue, }) require.NoError(t, err) + claimedJobs := claimRes.Jobs require.Len(t, claimedJobs, 1) require.Equal(t, job.ID, claimedJobs[0].ID) require.Equal(t, rivertype.JobStateRunning, claimedJobs[0].State) @@ -710,6 +712,67 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT require.False(t, gjson.GetBytes(updatedJob3.Metadata, "unique_key_conflict").Exists()) }) + // SQLite only: jobs whose rows can't be decoded are scheduled (or + // discarded for a unique conflict) without failing the rest of the + // batch, and the undecodable values are left in place. + t.Run("UndecodableJobsScheduled", func(t *testing.T) { + t.Parallel() + + exec, bundle := setup(ctx, t) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("only SQLite's JSON columns can hold values that don't decode") + } + + var ( + horizon = time.Now() + beforeHorizon = horizon.Add(-1 * time.Minute) + uniqueStates = uniquestates.UniqueStatesToBitmask([]rivertype.JobState{rivertype.JobStateAvailable, rivertype.JobStateRunning}) + ) + + var ( + goodJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ScheduledAt: &beforeHorizon, State: new(rivertype.JobStateRetryable)}) + undecodableJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ScheduledAt: &beforeHorizon, State: new(rivertype.JobStateRetryable)}) + conflictingJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ScheduledAt: &beforeHorizon, State: new(rivertype.JobStateRetryable), UniqueKey: []byte("unique-key"), UniqueStates: uniqueStates}) + scheduledJob = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ScheduledAt: &beforeHorizon, State: new(rivertype.JobStateScheduled)}) + undecodableValue = `{"not":"an array"}` + expectedJobIDs = []int64{goodJob.ID, undecodableJob.ID, conflictingJob.ID, scheduledJob.ID} + undecodableJobIDs = []int64{undecodableJob.ID, conflictingJob.ID, scheduledJob.ID} + ) + for _, jobID := range undecodableJobIDs { + sqliteSetJobJSONColumn(ctx, t, exec, jobID, "tags", undecodableValue) + } + + // Conflicts with conflictingJob, which is discarded instead of scheduled. + _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{ + State: new(rivertype.JobStateRunning), + UniqueKey: []byte("unique-key"), + UniqueStates: uniqueStates, + }) + + result, err := exec.JobSchedule(ctx, &riverdriver.JobScheduleParams{ + Max: 100, + Now: &horizon, + }) + require.NoError(t, err) + require.Equal(t, expectedJobIDs, + sliceutil.Map(result, func(r *riverdriver.JobScheduleResult) int64 { return r.Job.ID })) + require.Equal(t, []rivertype.JobState{rivertype.JobStateAvailable, rivertype.JobStateAvailable, rivertype.JobStateDiscarded, rivertype.JobStateAvailable}, + sliceutil.Map(result, func(r *riverdriver.JobScheduleResult) rivertype.JobState { return r.Job.State })) + require.Equal(t, []bool{false, false, true, false}, + sliceutil.Map(result, func(r *riverdriver.JobScheduleResult) bool { return r.ConflictDiscarded })) + + // Fields that could be decoded are set, while the others are empty. + require.Equal(t, []string{}, result[0].Job.Tags) + require.Nil(t, result[1].Job.Tags) + require.Equal(t, conflictingJob.Kind, result[2].Job.Kind) + + // The undecodable values are left as they were. + for _, jobID := range undecodableJobIDs { + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: jobID}) + require.ErrorContains(t, err, "error unmarshaling `tags`") + } + }) + t.Run("SchedulingTwoRetryableJobsThatWillConflictWithEachOther", func(t *testing.T) { t.Parallel() @@ -889,6 +952,29 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT require.Equal(t, "foo.go:123\nbar.go:456", jobAfter.Errors[0].Trace) }) + // SQLite only: a non-array `errors` value is wrapped in an array so that + // the new error can be appended without losing the existing value. + t.Run("NonArrayErrorsWrappedToAppend", func(t *testing.T) { + t.Parallel() + + exec, bundle := setup(ctx, t) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("only SQLite's JSON columns can hold a non-array errors value") + } + + now := precisionTestTime + + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: new(rivertype.JobStateRunning)}) + sqliteSetJobJSONColumn(ctx, t, exec, job.ID, "errors", `{"error":"existing value"}`) + + jobsAfter, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams(riverdriver.JobSetStateErrorRetryable(job.ID, now, makeErrPayload(t, now), nil))) + require.NoError(t, err) + require.Len(t, jobsAfter, 1) + require.Equal(t, rivertype.JobStateRetryable, jobsAfter[0].State) + require.Equal(t, []string{"existing value", "fake error"}, + sliceutil.Map(jobsAfter[0].Errors, func(e rivertype.AttemptError) string { return e.Error })) + }) + t.Run("SetsAnInterruptedRunningJobToAvailableWithUpdatedAttempt", func(t *testing.T) { t.Parallel() @@ -947,6 +1033,40 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT require.WithinDuration(t, job.ScheduledAt, jobAfter.ScheduledAt, time.Microsecond) }) + // A job whose row can't be fully decoded still has its state set, and + // doesn't prevent setting the state of other jobs in the same batch. + t.Run("UndecodableJobSetAlongsideOthers", func(t *testing.T) { + t.Parallel() + + exec, bundle := setup(ctx, t) + if bundle.driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("only SQLite's JSON columns can hold values that don't decode") + } + + now := precisionTestTime + + job1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: new(rivertype.JobStateRunning)}) + job2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: new(rivertype.JobStateRunning)}) + sqliteSetJobJSONColumn(ctx, t, exec, job2.ID, "tags", `{"not":"an array"}`) + + jobsAfter, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams( + riverdriver.JobSetStateErrorRetryable(job1.ID, now, makeErrPayload(t, now), nil), + riverdriver.JobSetStateErrorRetryable(job2.ID, now, makeErrPayload(t, now), nil), + )) + require.NoError(t, err) + require.Len(t, jobsAfter, 2) + for _, jobAfter := range jobsAfter { + require.Equal(t, rivertype.JobStateRetryable, jobAfter.State) + require.Len(t, jobAfter.Errors, 1) + } + require.Equal(t, job2.ID, jobsAfter[1].ID) + require.Nil(t, jobsAfter[1].Tags) + + // The undecodable value is left as it was. + _, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job2.ID}) + require.ErrorContains(t, err, "error unmarshaling `tags`") + }) + t.Run("UpdatesOnlyMetadataForAlreadyRetryableJobs", func(t *testing.T) { t.Parallel() diff --git a/riverdriver/riverdrivertest/riverdrivertest.go b/riverdriver/riverdrivertest/riverdrivertest.go index aa66762c..0f4e7086 100644 --- a/riverdriver/riverdrivertest/riverdrivertest.go +++ b/riverdriver/riverdrivertest/riverdrivertest.go @@ -109,3 +109,12 @@ func requireMissingRelation(t *testing.T, err error, schema, missingRelation str require.Regexp(t, fmt.Sprintf(`(pq: relation "%s\.%s" does not exist|no such table: %s\.%s|no such database: %s)`, schema, missingRelation, schema, missingRelation, schema), err.Error()) } } + +// sqliteSetJobJSONColumn overwrites a JSON column of a SQLite job row with the +// given JSON, simulating a row changed out of band into a shape that River +// can't decode. Postgres' column types don't allow the equivalent. +func sqliteSetJobJSONColumn(ctx context.Context, t *testing.T, exec riverdriver.Executor, jobID int64, column, jsonValue string) { + t.Helper() + + require.NoError(t, exec.Exec(ctx, "UPDATE river_job SET "+column+" = jsonb(?) WHERE id = ?", jsonValue, jobID)) +} diff --git a/riverdriver/riverpgxv5/river_pgx_v5_driver.go b/riverdriver/riverpgxv5/river_pgx_v5_driver.go index 62601230..66095b93 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/riverdriver/riverpgxv5/river_pgx_v5_driver.go @@ -316,7 +316,7 @@ func (e *Executor) JobDeleteMany(ctx context.Context, params *riverdriver.JobDel return sliceutil.MapError(jobs, jobRowFromInternal) } -func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { +func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { jobs, err := dbsqlc.New().JobGetAvailable(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetAvailableParams{ AttemptedBy: params.ClientID, MaxAttemptedBy: int32(min(params.MaxAttemptedBy, math.MaxInt32)), //nolint:gosec @@ -327,7 +327,7 @@ func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobG if err != nil { return nil, interpretError(err) } - return sliceutil.MapError(jobs, jobRowFromInternal) + return jobGetAvailableResultFromInternal(jobs), nil } func (e *Executor) JobGetByID(ctx context.Context, params *riverdriver.JobGetByIDParams) (*rivertype.JobRow, error) { @@ -363,7 +363,7 @@ func (e *Executor) JobGetStuck(ctx context.Context, params *riverdriver.JobGetSt if err != nil { return nil, interpretError(err) } - return sliceutil.MapError(jobs, jobRowFromInternal) + return jobRowsFromInternalPartial(jobs), nil } func (e *Executor) JobInsertFastMany(ctx context.Context, params *riverdriver.JobInsertFastManyParams) ([]*riverdriver.JobInsertFastResult, error) { @@ -670,7 +670,7 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr if err != nil { return nil, interpretError(err) } - return sliceutil.MapError(jobs, jobRowFromInternal) + return jobRowsFromInternalPartial(jobs), nil } func (e *Executor) JobUpdate(ctx context.Context, params *riverdriver.JobUpdateParams) (*rivertype.JobRow, error) { @@ -1239,17 +1239,49 @@ func interpretError(err error) error { return err } +// jobGetAvailableResultFromInternal decodes the job rows locked by +// JobGetAvailable, separating out any that can't be decoded rather than +// failing all of them, because they've all been moved to `running`. +func jobGetAvailableResultFromInternal(jobs []*dbsqlc.RiverJob) *riverdriver.JobGetAvailableResult { + res := &riverdriver.JobGetAvailableResult{Jobs: make([]*rivertype.JobRow, 0, len(jobs))} + for _, internal := range jobs { + job, err := jobRowFromInternalPartial(internal) + if err != nil { + res.UndecodableJobs = append(res.UndecodableJobs, &riverdriver.UndecodableJob{DecodeErr: err, Job: job}) + continue + } + res.Jobs = append(res.Jobs, job) + } + return res +} + +// jobRowFromInternal decodes a job row, returning an error if any of its +// fields can't be decoded. func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { + job, err := jobRowFromInternalPartial(internal) + if err != nil { + return nil, err + } + return job, nil +} + +// jobRowFromInternalPartial decodes a job row. A row is always returned, even +// along with an error, in which case the fields that couldn't be decoded are +// left empty. +func jobRowFromInternalPartial(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { var attemptedAt *time.Time if internal.AttemptedAt != nil { t := internal.AttemptedAt.UTC() attemptedAt = &t } + var decodeErr error errors := make([]rivertype.AttemptError, len(internal.Errors)) for i, rawError := range internal.Errors { if err := json.Unmarshal(rawError, &errors[i]); err != nil { - return nil, err + decodeErr = fmt.Errorf("error unmarshaling `errors`: %w", err) + errors = nil + break } } @@ -1283,7 +1315,17 @@ func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { Tags: internal.Tags, UniqueKey: internal.UniqueKey, UniqueStates: uniquestates.UniqueBitmaskToStates(uniqueStatesByte), - }, nil + }, decodeErr +} + +// jobRowsFromInternalPartial decodes job rows with jobRowFromInternalPartial, +// ignoring decode errors so that one bad row doesn't prevent returning the +// others. +func jobRowsFromInternalPartial(jobs []*dbsqlc.RiverJob) []*rivertype.JobRow { + return sliceutil.Map(jobs, func(internal *dbsqlc.RiverJob) *rivertype.JobRow { + job, _ := jobRowFromInternalPartial(internal) + return job + }) } func leaderFromInternal(internal *dbsqlc.RiverLeader) *riverdriver.Leader { diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index 528de8d8..3da03f63 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -603,7 +603,12 @@ SET attempt = CASE WHEN /* NOT should_cancel */(cast(@state AS text) <> 'available' AND @state <> 'retryable' AND @state <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(@attempt_do_update AS boolean) THEN @attempt ELSE attempt END, - errors = CASE WHEN cast(@errors_do_update AS boolean) + -- The errors column is always an array unless it's been changed out of + -- band. If it has, wrap its value in an array so that the new error is + -- still appended without losing it. + errors = CASE WHEN cast(@errors_do_update AS boolean) AND coalesce(json_type(errors), 'array') <> 'array' + THEN jsonb(json_array(json(errors), json(@error))) + WHEN cast(@errors_do_update AS boolean) THEN jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(@error))) ELSE errors END, finalized_at = CASE WHEN /* should_cancel */((@state = 'available' OR @state = 'retryable' OR @state = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL) diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go index 1bccd481..c13d1924 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -1643,7 +1643,12 @@ SET attempt = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'available' AND ?1 <> 'retryable' AND ?1 <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(?2 AS boolean) THEN ?3 ELSE attempt END, - errors = CASE WHEN cast(?4 AS boolean) + -- The errors column is always an array unless it's been changed out of + -- band. If it has, wrap its value in an array so that the new error is + -- still appended without losing it. + errors = CASE WHEN cast(?4 AS boolean) AND coalesce(json_type(errors), 'array') <> 'array' + THEN jsonb(json_array(json(errors), json(?5))) + WHEN cast(?4 AS boolean) THEN jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(?5))) ELSE errors END, finalized_at = CASE WHEN /* should_cancel */((?1 = 'available' OR ?1 = 'retryable' OR ?1 = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL) diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index a6b0264d..75e9d0ac 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -515,7 +515,7 @@ var jobGetAvailableAttemptedBySQL = strings.TrimSpace(` )) `) -func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { +func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { ctx = sqlctemplate.WithReplacements(ctx, map[string]sqlctemplate.Replacement{ "attempted_by_clause": { Stable: true, // input never changes @@ -534,7 +534,7 @@ func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobG if err != nil { return nil, interpretError(err) } - return sliceutil.MapError(jobs, jobRowFromInternal) + return jobGetAvailableResultFromInternal(jobs), nil } func (e *Executor) JobGetByID(ctx context.Context, params *riverdriver.JobGetByIDParams) (*rivertype.JobRow, error) { @@ -570,7 +570,7 @@ func (e *Executor) JobGetStuck(ctx context.Context, params *riverdriver.JobGetSt if err != nil { return nil, interpretError(err) } - return sliceutil.MapError(jobs, jobRowFromInternal) + return jobRowsFromInternalPartial(jobs), nil } func (e *Executor) JobInsertFastMany(ctx context.Context, params *riverdriver.JobInsertFastManyParams) ([]*riverdriver.JobInsertFastResult, error) { @@ -818,6 +818,15 @@ func (e *Executor) JobSchedule(ctx context.Context, params *riverdriver.JobSched scheduledResMap = make(map[int64]*riverdriver.JobScheduleResult) ) + // A job whose row can't be fully decoded is still returned (with the + // undecodable fields left empty) so that it doesn't fail scheduling + // for every other job. A job whose attempt fails because its row + // can't be decoded is retried, so its row comes through here. + scheduledJobRow := func(internal *dbsqlc.RiverJob) *rivertype.JobRow { + job, _ := jobRowFromInternalPartial(internal) + return job + } + for _, eligibleJob := range eligibleJobs { if eligibleJob.UniqueKey == nil { nonUniqueIDs = append(nonUniqueIDs, eligibleJob.ID) @@ -846,10 +855,7 @@ func (e *Executor) JobSchedule(ctx context.Context, params *riverdriver.JobSched if err != nil { return nil, interpretError(err) } - updatedJob, err := jobRowFromInternal(updatedJobs[0]) - if err != nil { - return nil, err - } + updatedJob := scheduledJobRow(updatedJobs[0]) scheduledResMap[updatedJob.ID] = &riverdriver.JobScheduleResult{Job: *updatedJob} } @@ -863,10 +869,7 @@ func (e *Executor) JobSchedule(ctx context.Context, params *riverdriver.JobSched } for _, internal := range updatedJobs { - updatedJob, err := jobRowFromInternal(internal) - if err != nil { - return nil, err - } + updatedJob := scheduledJobRow(internal) scheduledResMap[updatedJob.ID] = &riverdriver.JobScheduleResult{ConflictDiscarded: true, Job: *updatedJob} } } @@ -878,10 +881,7 @@ func (e *Executor) JobSchedule(ctx context.Context, params *riverdriver.JobSched } for _, internal := range updatedJobs { - updatedJob, err := jobRowFromInternal(internal) - if err != nil { - return nil, err - } + updatedJob := scheduledJobRow(internal) scheduledResMap[updatedJob.ID] = &riverdriver.JobScheduleResult{Job: *updatedJob} } } @@ -950,10 +950,10 @@ func (e *Executor) JobSetStateIfRunningMany(ctx context.Context, params *riverdr return fmt.Errorf("error setting job state: %w", err) } } - jobRow, err := jobRowFromInternal(job) - if err != nil { - return err - } + // A job whose row can't be fully decoded is still returned (with + // the undecodable fields left empty) so that it doesn't roll back + // the state change for every other job in the batch. + jobRow, _ := jobRowFromInternalPartial(job) setRes = append(setRes, jobRow) } @@ -1613,7 +1613,38 @@ func sqliteJobInsertFullManyJobsParam(jobs []*riverdriver.JobInsertFullParams) ( return json.Marshal(jobsParam) } +// jobGetAvailableResultFromInternal decodes the job rows locked by +// JobGetAvailable, separating out any that can't be decoded rather than +// failing all of them, because they've all been moved to `running`. +func jobGetAvailableResultFromInternal(jobs []*dbsqlc.RiverJob) *riverdriver.JobGetAvailableResult { + res := &riverdriver.JobGetAvailableResult{Jobs: make([]*rivertype.JobRow, 0, len(jobs))} + for _, internal := range jobs { + job, err := jobRowFromInternalPartial(internal) + if err != nil { + res.UndecodableJobs = append(res.UndecodableJobs, &riverdriver.UndecodableJob{DecodeErr: err, Job: job}) + continue + } + res.Jobs = append(res.Jobs, job) + } + return res +} + +// jobRowFromInternal decodes a job row, returning an error if any of its +// fields can't be decoded. func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { + job, err := jobRowFromInternalPartial(internal) + if err != nil { + return nil, err + } + return job, nil +} + +// jobRowFromInternalPartial decodes a job row. A row is always returned, even +// along with an error, in which case the fields that couldn't be decoded are +// left empty. +func jobRowFromInternalPartial(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { + var decodeErrs []error + var attemptedAt *time.Time if internal.AttemptedAt != nil { t := internal.AttemptedAt.UTC() @@ -1623,14 +1654,16 @@ func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { var attemptedBy []string if internal.AttemptedBy != nil { if err := json.Unmarshal(internal.AttemptedBy, &attemptedBy); err != nil { - return nil, fmt.Errorf("error unmarshaling `attempted_by`: %w", err) + decodeErrs = append(decodeErrs, fmt.Errorf("error unmarshaling `attempted_by`: %w", err)) + attemptedBy = nil } } - errors := make([]rivertype.AttemptError, 0) + attemptErrors := make([]rivertype.AttemptError, 0) if internal.Errors != nil { - if err := json.Unmarshal(internal.Errors, &errors); err != nil { - return nil, fmt.Errorf("error unmarshaling `errors`: %w", err) + if err := json.Unmarshal(internal.Errors, &attemptErrors); err != nil { + decodeErrs = append(decodeErrs, fmt.Errorf("error unmarshaling `errors`: %w", err)) + attemptErrors = nil } } @@ -1642,15 +1675,17 @@ func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { var tags []string if err := json.Unmarshal(internal.Tags, &tags); err != nil { - return nil, fmt.Errorf("error unmarshaling `tags`: %w", err) + decodeErrs = append(decodeErrs, fmt.Errorf("error unmarshaling `tags`: %w", err)) + tags = nil } var uniqueStatesByte byte if internal.UniqueStates != nil { if *internal.UniqueStates < 0 || *internal.UniqueStates > 255 { - return nil, fmt.Errorf("value out of range for byte: %d", *internal.UniqueStates) + decodeErrs = append(decodeErrs, fmt.Errorf("value out of range for byte: %d", *internal.UniqueStates)) + } else { + uniqueStatesByte = byte(*internal.UniqueStates) } - uniqueStatesByte = byte(*internal.UniqueStates) } return &rivertype.JobRow{ @@ -1660,7 +1695,7 @@ func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { AttemptedBy: attemptedBy, CreatedAt: internal.CreatedAt.UTC(), EncodedArgs: internal.Args, - Errors: errors, + Errors: attemptErrors, FinalizedAt: finalizedAt, Kind: internal.Kind, MaxAttempts: max(int(internal.MaxAttempts), 0), @@ -1672,7 +1707,17 @@ func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) { Tags: tags, UniqueKey: internal.UniqueKey, UniqueStates: uniquestates.UniqueBitmaskToStates(uniqueStatesByte), - }, nil + }, errors.Join(decodeErrs...) +} + +// jobRowsFromInternalPartial decodes job rows with jobRowFromInternalPartial, +// ignoring decode errors so that one bad row doesn't prevent returning the +// others. +func jobRowsFromInternalPartial(jobs []*dbsqlc.RiverJob) []*rivertype.JobRow { + return sliceutil.Map(jobs, func(internal *dbsqlc.RiverJob) *rivertype.JobRow { + job, _ := jobRowFromInternalPartial(internal) + return job + }) } func leaderFromInternal(internal *dbsqlc.RiverLeader) *riverdriver.Leader { diff --git a/rivershared/riverpilot/pilot.go b/rivershared/riverpilot/pilot.go index 69d48d3b..04f667f8 100644 --- a/rivershared/riverpilot/pilot.go +++ b/rivershared/riverpilot/pilot.go @@ -28,12 +28,15 @@ type Pilot interface { // conceivably be changed.) JobCleanerQueuesExcluded() []string + // JobGetAvailable locks available jobs for work. Locked jobs whose rows + // couldn't be fully decoded are returned in the result's UndecodableJobs + // and should have their attempt failed by the caller. JobGetAvailable( ctx context.Context, exec riverdriver.Executor, state ProducerState, params *riverdriver.JobGetAvailableParams, - ) ([]*rivertype.JobRow, error) + ) (*riverdriver.JobGetAvailableResult, error) JobInsertMany( ctx context.Context, diff --git a/rivershared/riverpilot/standard_pilot.go b/rivershared/riverpilot/standard_pilot.go index af2a92fe..12cf4ee2 100644 --- a/rivershared/riverpilot/standard_pilot.go +++ b/rivershared/riverpilot/standard_pilot.go @@ -17,12 +17,12 @@ type StandardPilot struct { func (p *StandardPilot) JobCleanerQueuesExcluded() []string { return nil } -func (p *StandardPilot) JobGetAvailable(ctx context.Context, exec riverdriver.Executor, state ProducerState, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { +func (p *StandardPilot) JobGetAvailable(ctx context.Context, exec riverdriver.Executor, state ProducerState, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { if params.MaxToLock <= 0 { - return nil, nil + return &riverdriver.JobGetAvailableResult{}, nil } - return timeoututil.WithTimeoutV(ctx, rivercommon.HotOperationTimeout, "StandardPilot.JobGetAvailable", func(ctx context.Context) ([]*rivertype.JobRow, error) { + return timeoututil.WithTimeoutV(ctx, rivercommon.HotOperationTimeout, "StandardPilot.JobGetAvailable", func(ctx context.Context) (*riverdriver.JobGetAvailableResult, error) { return exec.JobGetAvailable(ctx, params) }) } diff --git a/rivershared/riverpilot/standard_pilot_test.go b/rivershared/riverpilot/standard_pilot_test.go index ae7d4254..fc1c350f 100644 --- a/rivershared/riverpilot/standard_pilot_test.go +++ b/rivershared/riverpilot/standard_pilot_test.go @@ -8,16 +8,15 @@ import ( "github.com/stretchr/testify/require" "github.com/riverqueue/river/riverdriver" - "github.com/riverqueue/river/rivertype" ) type standardPilotExecutorMock struct { riverdriver.Executor - jobGetAvailableFunc func(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) + jobGetAvailableFunc func(ctx context.Context, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) } -func (m *standardPilotExecutorMock) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { +func (m *standardPilotExecutorMock) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { return m.jobGetAvailableFunc(ctx, params) } @@ -38,14 +37,14 @@ func TestStandardPilot_JobGetAvailable(t *testing.T) { } } - t.Run("ReturnsNilWhenMaxToLockIsZero", func(t *testing.T) { + t.Run("ReturnsEmptyWhenMaxToLockIsZero", func(t *testing.T) { t.Parallel() bundle := setup(t) res, err := bundle.pilot.JobGetAvailable(context.Background(), bundle.exec, nil, &riverdriver.JobGetAvailableParams{}) require.NoError(t, err) - require.Nil(t, res) + require.Equal(t, &riverdriver.JobGetAvailableResult{}, res) }) t.Run("PreservesParentCancellation", func(t *testing.T) { @@ -56,7 +55,7 @@ func TestStandardPilot_JobGetAvailable(t *testing.T) { parentCtx, cancel := context.WithCancelCause(context.Background()) cancel(parentErr) - bundle.exec.jobGetAvailableFunc = func(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { + bundle.exec.jobGetAvailableFunc = func(ctx context.Context, params *riverdriver.JobGetAvailableParams) (*riverdriver.JobGetAvailableResult, error) { <-ctx.Done() return nil, context.Cause(ctx) } diff --git a/rivertype/river_type.go b/rivertype/river_type.go index fe74d707..b8e7927b 100644 --- a/rivertype/river_type.go +++ b/rivertype/river_type.go @@ -4,9 +4,12 @@ package rivertype import ( + "bytes" "context" "encoding/json" "errors" + "math" + "strings" "time" ) @@ -280,6 +283,135 @@ type AttemptError struct { Trace string `json:"trace"` } +// UnmarshalJSON decodes an attempt error. River always persists attempt errors +// in the shape produced by encoding this type with encoding/json, and those +// decode exactly as they would with encoding/json's defaults. Elements written +// by other tools or edited by hand might not match that shape though, and +// because a job row can't be read or worked unless all of its attempt errors +// decode, any element that's valid JSON is decoded on a best effort basis +// instead of producing an error: +// +// - `at` accepts RFC 3339 timestamps, along with timestamps that use a +// space instead of `T`, a numeric UTC offset without a colon or minutes +// (as in Postgres' text output), or no offset at all (taken to be UTC). +// Any other value leaves At as the zero time. +// - `attempt` accepts integers, numbers with an integral value, and strings +// containing an integer. Any other value leaves Attempt as zero. +// - `error` and `trace` accept strings. Any other non-null value is kept as +// its JSON text. +// - An element that's a JSON string instead of an object is used as Error. +// Any other element that isn't an object is kept as its JSON text in +// Error. +// +// Only data that isn't valid JSON returns an error. +func (e *AttemptError) UnmarshalJSON(data []byte) error { + // Fast path for the common case where the element has the expected shape. + // An alias type without this method gets encoding/json's default behavior. + type attemptErrorAlias AttemptError + if err := json.Unmarshal(data, (*attemptErrorAlias)(e)); err == nil { + return nil + } + + if !json.Valid(data) { + return errors.New("attempt error is not valid JSON") + } + + data = bytes.TrimSpace(data) + if data[0] != '{' { + // Valid JSON, but not an object. + *e = AttemptError{Error: attemptErrorLenientString(data)} + return nil + } + + var fields struct { + At json.RawMessage `json:"at"` + Attempt json.RawMessage `json:"attempt"` + Error json.RawMessage `json:"error"` + Trace json.RawMessage `json:"trace"` + } + if err := json.Unmarshal(data, &fields); err != nil { + return err + } + + *e = AttemptError{ + At: attemptErrorLenientTime(fields.At), + Attempt: attemptErrorLenientInt(fields.Attempt), + Error: attemptErrorLenientString(fields.Error), + Trace: attemptErrorLenientString(fields.Trace), + } + return nil +} + +// Layouts accepted for AttemptError.At in addition to RFC 3339. A fractional +// second is optional in all of them. +// +//nolint:gochecknoglobals +var attemptErrorTimeLayouts = []string{ + "2006-01-02T15:04:05.999999999Z07:00", + "2006-01-02T15:04:05.999999999Z0700", + "2006-01-02T15:04:05.999999999Z07", + "2006-01-02T15:04:05.999999999", + "2006-01-02 15:04:05.999999999Z07:00", + "2006-01-02 15:04:05.999999999Z0700", + "2006-01-02 15:04:05.999999999Z07", + "2006-01-02 15:04:05.999999999", +} + +func attemptErrorLenientInt(data json.RawMessage) int { + var num json.Number + if err := json.Unmarshal(data, &num); err != nil { + var str string + if err := json.Unmarshal(data, &str); err != nil { + return 0 + } + num = json.Number(strings.TrimSpace(str)) + } + + if i, err := num.Int64(); err == nil { + return int(i) + } + // Floats are only accepted when they represent an integer exactly. + if f, err := num.Float64(); err == nil && f == math.Trunc(f) && math.Abs(f) <= 1<<53 { + return int(f) + } + return 0 +} + +func attemptErrorLenientString(data json.RawMessage) string { + if len(data) == 0 { + return "" + } + + var str *string + if err := json.Unmarshal(data, &str); err == nil { + if str == nil { // JSON null + return "" + } + return *str + } + + var compacted bytes.Buffer + if err := json.Compact(&compacted, data); err != nil { + return string(data) + } + return compacted.String() +} + +func attemptErrorLenientTime(data json.RawMessage) time.Time { + var str string + if err := json.Unmarshal(data, &str); err != nil { + return time.Time{} + } + + str = strings.TrimSpace(str) + for _, layout := range attemptErrorTimeLayouts { + if t, err := time.Parse(layout, str); err == nil { + return t + } + } + return time.Time{} +} + type JobInsertParams struct { ID *int64 Args JobArgs diff --git a/rivertype/river_type_test.go b/rivertype/river_type_test.go index aa677ac8..e8019fac 100644 --- a/rivertype/river_type_test.go +++ b/rivertype/river_type_test.go @@ -1,17 +1,115 @@ package rivertype_test import ( + "encoding/json" "go/ast" "go/parser" "go/token" "os" "testing" + "time" "github.com/stretchr/testify/require" "github.com/riverqueue/river/rivertype" ) +func TestAttemptError_UnmarshalJSON(t *testing.T) { + t.Parallel() + + attemptAt := time.Date(2024, 1, 2, 3, 4, 5, 123456000, time.UTC) + + t.Run("InvalidJSON", func(t *testing.T) { + t.Parallel() + + var attemptErr rivertype.AttemptError + require.EqualError(t, attemptErr.UnmarshalJSON([]byte(`{"at":`)), "attempt error is not valid JSON") + }) + + t.Run("Lenient", func(t *testing.T) { + t.Parallel() + + tests := []struct { + expected rivertype.AttemptError + json string + name string + }{ + {name: "AtInvalid", json: `{"at":"not a time","attempt":2,"error":"err"}`, expected: rivertype.AttemptError{Attempt: 2, Error: "err"}}, + {name: "AtNoOffset", json: `{"at":"2024-01-02T03:04:05.123456","attempt":2}`, expected: rivertype.AttemptError{At: attemptAt, Attempt: 2}}, + {name: "AtNumber", json: `{"at":1704164645,"attempt":2}`, expected: rivertype.AttemptError{Attempt: 2}}, + {name: "AtPostgresText", json: `{"at":"2024-01-02 03:04:05.123456+00","attempt":2}`, expected: rivertype.AttemptError{At: attemptAt, Attempt: 2}}, + {name: "AtSpaceNoOffset", json: `{"at":"2024-01-02 03:04:05.123456","attempt":2}`, expected: rivertype.AttemptError{At: attemptAt, Attempt: 2}}, + {name: "AttemptFloat", json: `{"attempt":3.0,"error":"err"}`, expected: rivertype.AttemptError{Attempt: 3, Error: "err"}}, + {name: "AttemptFractional", json: `{"attempt":3.5,"error":"err"}`, expected: rivertype.AttemptError{Error: "err"}}, + {name: "AttemptObject", json: `{"attempt":{},"error":"err"}`, expected: rivertype.AttemptError{Error: "err"}}, + {name: "AttemptString", json: `{"attempt":" 3 ","error":"err"}`, expected: rivertype.AttemptError{Attempt: 3, Error: "err"}}, + {name: "AttemptStringInvalid", json: `{"attempt":"three","error":"err"}`, expected: rivertype.AttemptError{Error: "err"}}, + {name: "ElementArray", json: `[1, "two"]`, expected: rivertype.AttemptError{Error: `[1,"two"]`}}, + {name: "ElementNumber", json: `123`, expected: rivertype.AttemptError{Error: "123"}}, + {name: "ElementString", json: `"job failed"`, expected: rivertype.AttemptError{Error: "job failed"}}, + {name: "ErrorObject", json: `{"attempt":1,"error":{"message": "boom", "code": 7}}`, expected: rivertype.AttemptError{Attempt: 1, Error: `{"message":"boom","code":7}`}}, + {name: "TraceArray", json: `{"attempt":1,"error":"err","trace":["frame1", "frame2"]}`, expected: rivertype.AttemptError{Attempt: 1, Error: "err", Trace: `["frame1","frame2"]`}}, + {name: "TraceNullWithInvalidField", json: `{"attempt":"x","error":null,"trace":null}`, expected: rivertype.AttemptError{}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + var attemptErr rivertype.AttemptError + require.NoError(t, json.Unmarshal([]byte(test.json), &attemptErr)) + require.True(t, test.expected.At.Equal(attemptErr.At), "expected at %s, got %s", test.expected.At, attemptErr.At) + attemptErr.At = test.expected.At + require.Equal(t, test.expected, attemptErr) + }) + } + }) + + t.Run("RoundTrip", func(t *testing.T) { + t.Parallel() + + attemptErr := rivertype.AttemptError{ + At: attemptAt, + Attempt: 3, + Error: "job failed", + Trace: "goroutine 1 [running]:", + } + data, err := json.Marshal(attemptErr) + require.NoError(t, err) + require.JSONEq(t, `{"at":"2024-01-02T03:04:05.123456Z","attempt":3,"error":"job failed","trace":"goroutine 1 [running]:"}`, string(data)) + + var decoded rivertype.AttemptError + require.NoError(t, json.Unmarshal(data, &decoded)) + require.Equal(t, attemptErr, decoded) + }) + + t.Run("Slice", func(t *testing.T) { + t.Parallel() + + // One unexpected element doesn't prevent decoding the others. + var attemptErrs []rivertype.AttemptError + require.NoError(t, json.Unmarshal([]byte(`[{"at":"2024-01-02T03:04:05.123456Z","attempt":1,"error":"err1","trace":""},"err2"]`), &attemptErrs)) + require.Equal(t, []rivertype.AttemptError{ + {At: attemptAt, Attempt: 1, Error: "err1"}, + {Error: "err2"}, + }, attemptErrs) + }) + + t.Run("StrictShapeUnchanged", func(t *testing.T) { + t.Parallel() + + // Missing fields and nulls decode the same as with encoding/json's + // defaults. + var attemptErr rivertype.AttemptError + require.NoError(t, json.Unmarshal([]byte(`{"attempt":2,"error":null}`), &attemptErr)) + require.Equal(t, rivertype.AttemptError{Attempt: 2}, attemptErr) + + attemptErr = rivertype.AttemptError{Error: "previous"} + require.NoError(t, json.Unmarshal([]byte(`null`), &attemptErr)) + require.Equal(t, rivertype.AttemptError{Error: "previous"}, attemptErr) + }) +} + func TestJobRow_Output(t *testing.T) { t.Parallel()