diff --git a/CHANGELOG.md b/CHANGELOG.md index 35745b30..0fa5db07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Guard against empty job slice returned by `JobSetStateIfRunningMany` when a job has been deleted mid-run. [PR #1308](https://github.com/riverqueue/river/pull/1308). - Fixed `JobRescuer` pagination so a full batch of running jobs with disabled or longer worker-specific timeouts can't prevent later stuck jobs from being rescued. [PR #1318](https://github.com/riverqueue/river/pull/1318). +- If a job fails to unmarshal from JSON during job rescue or job execution, back off using the retry schedule and eventually discard it, similar to any other error that might occur. [PR #1324](https://github.com/riverqueue/river/pull/1324). ## [0.40.0] - 2026-07-02 diff --git a/internal/jobexecutor/job_executor.go b/internal/jobexecutor/job_executor.go index ae7cab09..97c5beb5 100644 --- a/internal/jobexecutor/job_executor.go +++ b/internal/jobexecutor/job_executor.go @@ -80,11 +80,12 @@ func MetadataUpdatesFromWorkContext(ctx context.Context) (map[string]any, bool) } type jobExecutorResult struct { - Err error - MetadataUpdates map[string]any - NextRetry time.Time - PanicTrace string - PanicVal any + Err error + JobArgsUnmarshaled bool + MetadataUpdates map[string]any + NextRetry time.Time + PanicTrace string + PanicVal any } // ErrorStr returns an appropriate string to persist to the database based on @@ -183,6 +184,7 @@ func (e *JobExecutor) Execute(ctx context.Context) { func (e *JobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { metadataUpdates := make(map[string]any) ctx = context.WithValue(ctx, ContextKeyMetadataUpdates, metadataUpdates) + jobArgsUnmarshaled := false defer func() { if recovery := recover(); recovery != nil { @@ -193,7 +195,8 @@ func (e *JobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { ) res = &jobExecutorResult{ - MetadataUpdates: metadataUpdates, + JobArgsUnmarshaled: jobArgsUnmarshaled, + MetadataUpdates: metadataUpdates, // Skip the first 4 frames which are: // // 1. The `runtime.Callers` function. @@ -230,6 +233,7 @@ func (e *JobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { if err := e.WorkUnit.UnmarshalJob(); err != nil { return err } + jobArgsUnmarshaled = true jobTimeout := cmp.Or(e.WorkUnit.Timeout(), e.ClientJobTimeout) @@ -268,7 +272,11 @@ func (e *JobExecutor) execute(ctx context.Context) (res *jobExecutorResult) { e.JobRow, ) - return &jobExecutorResult{Err: executeFunc(ctx), MetadataUpdates: metadataUpdates} + return &jobExecutorResult{ + Err: executeFunc(ctx), + JobArgsUnmarshaled: jobArgsUnmarshaled, + MetadataUpdates: metadataUpdates, + } } // Watches for jobs that may have become stuck. i.e. They've run longer than @@ -491,7 +499,7 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow, } var nextRetryScheduledAt time.Time - if e.WorkUnit != nil { + if e.WorkUnit != nil && res.JobArgsUnmarshaled { nextRetryScheduledAt = e.WorkUnit.NextRetry() } if nextRetryScheduledAt.IsZero() { diff --git a/internal/jobexecutor/job_executor_test.go b/internal/jobexecutor/job_executor_test.go index c2925379..f595481b 100644 --- a/internal/jobexecutor/job_executor_test.go +++ b/internal/jobexecutor/job_executor_test.go @@ -31,10 +31,11 @@ import ( // of the workUnit. Unlike in other packages, this one does not make use of any // types from the top level river package (like `river.Job[T]`). type customizableWorkUnit struct { - middleware []rivertype.WorkerMiddleware - nextRetry func() time.Time - timeout time.Duration - work func() error + middleware []rivertype.WorkerMiddleware + nextRetry func() time.Time + timeout time.Duration + unmarshalErr error + work func() error } func (w *customizableWorkUnit) PluginLookup(lookup *pluginlookup.JobPluginLookup) pluginlookup.PluginLookupInterface { @@ -57,7 +58,7 @@ func (w *customizableWorkUnit) Timeout() time.Duration { } func (w *customizableWorkUnit) UnmarshalJob() error { - return nil + return w.unmarshalErr } func (w *customizableWorkUnit) Work(ctx context.Context) error { @@ -488,6 +489,42 @@ func TestJobExecutor_Execute(t *testing.T) { require.WithinDuration(t, nextRetryAt, job.ScheduledAt, time.Microsecond) }) + t.Run("ErrorUnmarshalingUsesClientRetryPolicy", func(t *testing.T) { + t.Parallel() + + executor, bundle := setup(t) + executor.ClientRetryPolicy = &retrypolicytest.RetryPolicyCustom{} + + var nextRetryCalled, workCalled bool + unmarshalErr := errors.New("invalid job args") + executor.WorkUnit = &customizableWorkUnit{ + nextRetry: func() time.Time { + nextRetryCalled = true + return time.Now().Add(1 * time.Hour) + }, + unmarshalErr: unmarshalErr, + work: func() error { + workCalled = true + return nil + }, + } + expectedRetryAt := executor.ClientRetryPolicy.NextRetry(bundle.jobRow) + + 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.JobStateRetryable, job.State) + require.Equal(t, unmarshalErr.Error(), job.Errors[0].Error) + require.WithinDuration(t, expectedRetryAt, job.ScheduledAt, time.Microsecond) + require.False(t, nextRetryCalled) + require.False(t, workCalled) + }) + t.Run("InvalidNextRetryAt", func(t *testing.T) { t.Parallel() diff --git a/internal/maintenance/job_rescuer.go b/internal/maintenance/job_rescuer.go index 81a89a84..bb5b821f 100644 --- a/internal/maintenance/job_rescuer.go +++ b/internal/maintenance/job_rescuer.go @@ -327,8 +327,17 @@ func (s *JobRescuer) makeRetryDecision(ctx context.Context, job *rivertype.JobRo workUnit := workUnitFactory.MakeUnit(job) if err := workUnit.UnmarshalJob(); err != nil { - s.Logger.ErrorContext(ctx, s.Name+": Error unmarshaling job args: %s"+err.Error(), - slog.String("job_kind", job.Kind), slog.Int64("job_id", job.ID)) + s.Logger.ErrorContext(ctx, s.Name+": Error unmarshaling job args", + slog.String("error", err.Error()), + slog.String("job_kind", job.Kind), + slog.Int64("job_id", job.ID), + ) + + if job.Attempt < max(job.MaxAttempts, 0) { + return jobRetryDecisionRetry, s.Config.ClientRetryPolicy.NextRetry(job) + } + + return jobRetryDecisionDiscard, time.Time{} } timeout := workUnit.Timeout() diff --git a/internal/maintenance/job_rescuer_test.go b/internal/maintenance/job_rescuer_test.go index a5024a88..c6143c2e 100644 --- a/internal/maintenance/job_rescuer_test.go +++ b/internal/maintenance/job_rescuer_test.go @@ -2,6 +2,7 @@ package maintenance import ( "context" + "errors" "fmt" "math" "sync/atomic" @@ -27,29 +28,47 @@ import ( // callbackWorkUnitFactory wraps a Worker to implement workUnitFactory. type callbackWorkUnitFactory struct { - Callback func(ctx context.Context, jobRow *rivertype.JobRow) error - timeout time.Duration // defaults to 0, which signals default timeout + Callback func(ctx context.Context, jobRow *rivertype.JobRow) error + timeout time.Duration // defaults to 0, which signals default timeout + unmarshalErr error } func (w *callbackWorkUnitFactory) MakeUnit(jobRow *rivertype.JobRow) workunit.WorkUnit { - return &callbackWorkUnit{callback: w.Callback, jobRow: jobRow, timeout: w.timeout} + return &callbackWorkUnit{ + callback: w.Callback, + jobRow: jobRow, + timeout: w.timeout, + unmarshalErr: w.unmarshalErr, + } } // callbackWorkUnit implements workUnit for a job and Worker. type callbackWorkUnit struct { - callback func(ctx context.Context, jobRow *rivertype.JobRow) error - jobRow *rivertype.JobRow - timeout time.Duration // defaults to 0, which signals default timeout + callback func(ctx context.Context, jobRow *rivertype.JobRow) error + jobRow *rivertype.JobRow + timeout time.Duration // defaults to 0, which signals default timeout + unmarshalErr error } func (w *callbackWorkUnit) PluginLookup(cache *pluginlookup.JobPluginLookup) pluginlookup.PluginLookupInterface { return nil } func (w *callbackWorkUnit) Middleware() []rivertype.WorkerMiddleware { return nil } -func (w *callbackWorkUnit) NextRetry() time.Time { return time.Now().Add(30 * time.Second) } -func (w *callbackWorkUnit) Timeout() time.Duration { return w.timeout } -func (w *callbackWorkUnit) Work(ctx context.Context) error { return w.callback(ctx, w.jobRow) } -func (w *callbackWorkUnit) UnmarshalJob() error { return nil } +func (w *callbackWorkUnit) NextRetry() time.Time { + if w.unmarshalErr != nil { + panic("NextRetry must not be called after UnmarshalJob returns an error") + } + return time.Now().Add(30 * time.Second) +} + +func (w *callbackWorkUnit) Timeout() time.Duration { + if w.unmarshalErr != nil { + panic("Timeout must not be called after UnmarshalJob returns an error") + } + return w.timeout +} +func (w *callbackWorkUnit) Work(ctx context.Context) error { return w.callback(ctx, w.jobRow) } +func (w *callbackWorkUnit) UnmarshalJob() error { return w.unmarshalErr } type SimpleClientRetryPolicy struct{} @@ -362,6 +381,57 @@ func TestJobRescuer(t *testing.T) { riversharedtest.WaitOrTimeout(t, stopped) }) + t.Run("UnmarshalErrorDiscardsAtMaxAttempts", func(t *testing.T) { + t.Parallel() + + rescuer, _ := setup(t) + + rescuer.Config.WorkUnitFactoryFunc = func(kind string) workunit.WorkUnitFactory { + return &callbackWorkUnitFactory{ + unmarshalErr: errors.New("invalid job args"), + } + } + + attemptedAt := time.Now().Add(-2 * JobRescuerRescueAfterDefault) + job := &rivertype.JobRow{ + ID: 123, + Attempt: 5, + AttemptedAt: &attemptedAt, + Kind: rescuerJobKind, + MaxAttempts: 5, + } + + decision, retryAt := rescuer.makeRetryDecision(ctx, job, time.Now()) + require.Equal(t, jobRetryDecisionDiscard, decision) + require.Zero(t, retryAt) + }) + + t.Run("UnmarshalErrorRetriesWithClientPolicy", func(t *testing.T) { + t.Parallel() + + rescuer, _ := setup(t) + + rescuer.Config.WorkUnitFactoryFunc = func(kind string) workunit.WorkUnitFactory { + return &callbackWorkUnitFactory{ + unmarshalErr: errors.New("invalid job args"), + } + } + + attemptedAt := time.Now().Add(-2 * JobRescuerRescueAfterDefault) + job := &rivertype.JobRow{ + ID: 123, + Attempt: 1, + AttemptedAt: &attemptedAt, + Kind: rescuerJobKind, + MaxAttempts: 5, + } + expectedRetryAt := rescuer.Config.ClientRetryPolicy.NextRetry(job) + + decision, retryAt := rescuer.makeRetryDecision(ctx, job, time.Now()) + require.Equal(t, jobRetryDecisionRetry, decision) + require.Equal(t, expectedRetryAt, retryAt) + }) + t.Run("UsesPilot", func(t *testing.T) { t.Parallel()