Isolate fetched jobs whose rows can't be decoded

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
 

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

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

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

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 {

producer_test.go

@@ -3,6 +3,7 @@ package river
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"slices"
 	"sync/atomic"
@@ -45,14 +46,47 @@ 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)
 	}
 
 	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()