Harden JSON unmarshaling in job rescuer + job executor

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
 

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() {

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()
 

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()

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()