From ece9af64b3b459df91e4bf767e2774b3ab56f081 Mon Sep 17 00:00:00 2001 From: Brandur Date: Wed, 22 Jul 2026 13:13:06 -0500 Subject: [PATCH] Harden JSON unmarshaling in job rescuer + job executor Here, do a little more to harden unmarshaling in the job rescuer and job executor. Previously, if a job's JSON didn't unmarshal successfully, we'd log an error and move on, which could result in unexpected trouble. Realistically, this would be a very rare circumstance though luckily -- the Postgres/SQLite `jsonb` fields guarantee the formatting of the JSON data, so the only way for an unmarshaling problem to occur is if a job's data was stored initially, and then the JSON Go struct later change to be incompatible (i.e. string changed to an int or something of that nature). Here, during an unmarshal error, use the job's standard retry schedule and back off, discarding the job if it's at the end of its allowed retries. This approach is best because it gives the user a chance to notice the failure and correct a potential unmarshaling problem by either manipulating the job row's data or fixing their Go struct. Fixes #1323, but also addresses a similar problem in the job executor that Codex found while working. --- CHANGELOG.md | 1 + internal/jobexecutor/job_executor.go | 24 ++++-- internal/jobexecutor/job_executor_test.go | 47 ++++++++++-- internal/maintenance/job_rescuer.go | 13 +++- internal/maintenance/job_rescuer_test.go | 90 ++++++++++++++++++++--- 5 files changed, 150 insertions(+), 25 deletions(-) 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()