From 07273d52e2261647a4c9e6169eda4b361c7dcc9a Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Thu, 24 Sep 2026 12:42:58 -0500 Subject: [PATCH 1/4] decode attempt errors leniently Every element of a job's `errors` has to decode into `rivertype.AttemptError` for the job row to be readable. An element that's valid JSON but not in the shape River writes, like an `at` in Postgres' text timestamp format, an `attempt` stored as a string, or a structured `error` or `trace` written by another tool, fails the whole row. When the row is one of the jobs locked by a fetch, the fetch returns an error after the claim has already moved every locked job to `running`, so all of them sit there without being worked, and the rescuer fails on the same row when it tries to recover them. Give `AttemptError` an `UnmarshalJSON` that keeps encoding/json's behavior for well-formed elements and falls back to best effort decoding for any other valid JSON: `at` also accepts a few common timestamp variants and is otherwise left zero, `attempt` accepts integral numbers and numeric strings, non-string `error` and `trace` values are kept as their JSON text, and a non-object element becomes the error message. Only invalid JSON still returns an error, which a Postgres `jsonb[]` element can never be. The persisted format doesn't change. Unit tests cover each fallback, and a new driver test fetches a job with oddly shaped attempt errors alongside a normal one on every driver. --- CHANGELOG.md | 1 + riverdriver/riverdrivertest/job_read.go | 43 ++++++++ rivertype/river_type.go | 132 ++++++++++++++++++++++++ rivertype/river_type_test.go | 101 ++++++++++++++++++ 4 files changed, 277 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e0952855..710421e5f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed SQLite notification listeners delivering notifications from before a subscription or from an unsubscribe gap. Notification reads now fetch subscribed topics in bounded batches, and cleanup deletes expired notifications in batches of 10,000 rows (reduced to 1,000 after repeated timeouts), with pauses between batches to reduce write lock contention. [PR #1381](https://github.com/riverqueue/river/pull/1381). - Fixed SQLite reusing the ID of a deleted job when that job held the largest ID, which could cause an ID observed earlier to refer to an unrelated job later. [PR #1390](https://github.com/riverqueue/river/pull/1390). - Fixed the `Job appears to be stuck` log line reporting the client-level `JobTimeout` instead of the worker-level timeout when a worker overrides `Timeout`. [PR #1394](https://github.com/riverqueue/river/pull/1394). +- 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). ## [0.47.0] - 2026-09-01 diff --git a/riverdriver/riverdrivertest/job_read.go b/riverdriver/riverdrivertest/job_read.go index fc3abe8c7..132d2f48a 100644 --- a/riverdriver/riverdrivertest/job_read.go +++ b/riverdriver/riverdrivertest/job_read.go @@ -518,6 +518,49 @@ 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, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + ClientID: testClientID, + MaxAttemptedBy: maxAttemptedBy, + MaxToLock: maxToLock, + Queue: rivercommon.QueueDefault, + }) + require.NoError(t, err) + + // 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 + })) + }) }) t.Run("JobGetByID", func(t *testing.T) { diff --git a/rivertype/river_type.go b/rivertype/river_type.go index fe74d7076..b8e7927ba 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 aa677ac81..7ba5c380b 100644 --- a/rivertype/river_type_test.go +++ b/rivertype/river_type_test.go @@ -1,17 +1,118 @@ 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") + }) + + // Older or externally edited attempt errors can contain valid JSON with + // unexpected shapes. Keep every field that can still be interpreted so + // one historical error cannot make the whole job unreadable. + t.Run("UnexpectedShapesPreserveUsableFields", 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() From a2f3110722cfff290dbe220dd6c5c28d0e1fcac0 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Thu, 24 Sep 2026 12:56:50 -0500 Subject: [PATCH 2/4] fail undecodable fetched jobs instead of stalling The fetch query moves every job it locks to `running` before the driver decodes the returned rows, and one row that can't be decoded fails the whole fetch. The producer only logs that error, so every job locked alongside the bad one is left `running` without being worked. The rescuer can't recover them either, since it reads stuck jobs with the same strict decoding and fails on the same row. After lenient attempt error decoding this can no longer happen on Postgres, but on SQLite `errors`, `tags`, and `attempted_by` are JSON columns that can be changed to any shape. Have `JobGetAvailable` decode each locked row separately and return a `JobGetAvailableResult` with the jobs that decoded and, separately, the `UndecodableJobs` along with the fields that could be decoded and the decode error. The pilot passes the result through, and the producer starts an executor for each undecodable job with the decode error set. The executor doesn't work such a job. It fails the attempt with an error describing the decode failure before hooks or middleware run, the same way it handles an unknown job kind, so the error handler is invoked and the job is retried with the client's retry policy or discarded at its max attempts, emitting a failed event. The attempt error is appended in SQL without decoding the bad column, and the bad values are left as they were. On SQLite, an `errors` value that isn't an array is wrapped in one so the new error can still be appended. `JobSetStateIfRunningMany` and `JobGetStuck` return rows that can't be fully decoded with the undecodable fields left empty instead of failing, so completing an undecodable job doesn't fail (or on SQLite, roll back) the rest of its batch, and the rescuer can recover jobs stranded by this problem before upgrading. Driver tests cover each of these on SQLite, and client tests show good jobs completing alongside undecodable ones that end up `retryable` or `discarded` with the decode error. Producer and executor tests exercise the same path on Postgres using a pilot that reports a job as undecodable. --- CHANGELOG.md | 1 + client_pilot_test.go | 2 +- internal/jobexecutor/job_executor.go | 19 +- internal/jobexecutor/job_executor_test.go | 72 ++++++- producer.go | 46 +++-- producer_test.go | 96 ++++++++- riverdriver/river_driver_interface.go | 43 +++- .../river_database_sql_driver.go | 63 +++++- riverdriver/riverdrivertest/benchmark.go | 3 +- .../riverdrivertest/driver_client_test.go | 93 +++++++++ riverdriver/riverdrivertest/job_read.go | 187 ++++++++++++------ riverdriver/riverdrivertest/job_update.go | 61 +++++- .../riverdrivertest/riverdrivertest.go | 9 + riverdriver/riverpgxv5/river_pgx_v5_driver.go | 54 ++++- .../riversqlite/internal/dbsqlc/river_job.sql | 7 +- .../internal/dbsqlc/river_job.sql.go | 7 +- .../riversqlite/river_sqlite_driver.go | 77 ++++++-- rivershared/riverpilot/pilot.go | 5 +- rivershared/riverpilot/standard_pilot.go | 6 +- rivershared/riverpilot/standard_pilot_test.go | 11 +- 20 files changed, 729 insertions(+), 133 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 710421e5f..9349db6e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed SQLite reusing the ID of a deleted job when that job held the largest ID, which could cause an ID observed earlier to refer to an unrelated job later. [PR #1390](https://github.com/riverqueue/river/pull/1390). - Fixed the `Job appears to be stuck` log line reporting the client-level `JobTimeout` instead of the worker-level timeout when a worker overrides `Timeout`. [PR #1394](https://github.com/riverqueue/river/pull/1394). - 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 the rescuer no longer fails on stuck jobs that can't be decoded. [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 55e81c1a0..fc9bc160c 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 21b63c68d..7e9075f3e 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 0e37a2fcd..2b83f8226 100644 --- a/internal/jobexecutor/job_executor_test.go +++ b/internal/jobexecutor/job_executor_test.go @@ -161,13 +161,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] @@ -564,6 +565,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() @@ -678,8 +744,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 477ca62ee..81fe63194 100644 --- a/producer.go +++ b/producer.go @@ -686,10 +686,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, pendingCancelJobs) + } else if numLocked := len(result.jobs) + len(result.undecodableJobs); numLocked > 0 { + p.startNewExecutors(workCtx, result.jobs, result.undecodableJobs, pendingCancelJobs) - 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. @@ -884,7 +884,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, Kind: p.fetchKind, MaxAttemptedBy: maxAttemptedBy, @@ -898,6 +898,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{ @@ -908,13 +911,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) { @@ -959,15 +962,19 @@ func (p *producer) heartbeatLogLoop(ctx context.Context, wg *sync.WaitGroup) { } } -func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.JobRow, pendingCancelJobs map[int64]struct{}) { +// 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, pendingCancelJobs map[int64]struct{}) { 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. @@ -984,6 +991,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) @@ -1006,7 +1014,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{}{}) } @@ -1225,8 +1240,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 e4f9cd0fd..c6620a658 100644 --- a/producer_test.go +++ b/producer_test.go @@ -3,6 +3,7 @@ package river import ( "context" "encoding/json" + "errors" "fmt" "slices" "sync/atomic" @@ -38,7 +39,7 @@ const testClientID = "test-client-id" type afterJobGetAvailablePilot struct { riverpilot.Pilot - afterJobGetAvailableFunc func(jobs []*rivertype.JobRow, err error) + afterJobGetAvailableFunc func(res *riverdriver.JobGetAvailableResult, err error) } func (p *afterJobGetAvailablePilot) JobGetAvailable( @@ -46,12 +47,12 @@ func (p *afterJobGetAvailablePilot) JobGetAvailable( exec riverdriver.Executor, state riverpilot.ProducerState, params *riverdriver.JobGetAvailableParams, -) ([]*rivertype.JobRow, error) { - jobs, err := p.Pilot.JobGetAvailable(ctx, exec, state, params) +) (*riverdriver.JobGetAvailableResult, error) { + res, err := p.Pilot.JobGetAvailable(ctx, exec, state, params) if p.afterJobGetAvailableFunc != nil { - p.afterJobGetAvailableFunc(jobs, err) + p.afterJobGetAvailableFunc(res, err) } - return jobs, err + return res, err } // beforeJobGetAvailablePilot calls a hook before delegating JobGetAvailable to @@ -67,7 +68,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) } @@ -75,6 +76,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() @@ -548,6 +582,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() @@ -596,12 +674,12 @@ func testProducer(t *testing.T, makeProducer func(ctx context.Context, t *testin workerErr.Init(t) producer.pilot = &afterJobGetAvailablePilot{ Pilot: producer.pilot, - afterJobGetAvailableFunc: func(jobs []*rivertype.JobRow, err error) { - if err != nil || len(jobs) == 0 { + afterJobGetAvailableFunc: func(res *riverdriver.JobGetAvailableResult, err error) { + if err != nil || res == nil || len(res.Jobs) == 0 { fetchReturned.Signal(0) return } - fetchReturned.Signal(jobs[0].ID) + fetchReturned.Signal(res.Jobs[0].ID) <-releaseFetch.WaitC() }, } diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index f483c4499..f68f066e1 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -240,7 +240,14 @@ 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) @@ -249,7 +256,12 @@ type Executor interface { // restricted to the provided IDs. JobGetCancelRequested(ctx context.Context, params *JobGetCancelRequestedParams) ([]int64, 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) @@ -259,7 +271,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) @@ -455,6 +473,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 @@ -978,6 +1007,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 4ba696192..9b48f12a7 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/riverdriver/riverdatabasesql/river_database_sql_driver.go @@ -369,7 +369,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, Kind: params.Kind, @@ -381,7 +381,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) { @@ -422,7 +422,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) { @@ -777,7 +777,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) { @@ -1326,17 +1326,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 } } @@ -1348,7 +1381,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{ @@ -1358,7 +1391,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), @@ -1370,7 +1403,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 d0118d3e4..dde1e0862 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 a1b6af22a..b5871103c 100644 --- a/riverdriver/riverdrivertest/driver_client_test.go +++ b/riverdriver/riverdrivertest/driver_client_test.go @@ -277,6 +277,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]) { @@ -1486,6 +1494,32 @@ func ExerciseClient[TTx any](ctx context.Context, t *testing.T, } }) + // 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() @@ -1631,4 +1665,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 132d2f48a..8aaa1fefd 100644 --- a/riverdriver/riverdrivertest/job_read.go +++ b/riverdriver/riverdrivertest/job_read.go @@ -243,6 +243,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() @@ -250,13 +260,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] @@ -276,14 +285,13 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx // Filtering precedes the limit: the earlier, higher priority unknown // job must not prevent either known kind from being claimed. for _, expected := range []*rivertype.JobRow{known1, known2} { - jobs, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobs := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, Kind: []string{"known1", "known2"}, MaxAttemptedBy: maxAttemptedBy, MaxToLock: 1, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Len(t, jobs, 1) require.Equal(t, expected.ID, jobs[0].ID) require.Equal(t, expected.Attempt+1, jobs[0].Attempt) @@ -303,14 +311,13 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx exec, _ := setup(ctx, t) job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{}) - jobs, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{ + jobs := getAvailable(t, exec, &riverdriver.JobGetAvailableParams{ ClientID: testClientID, Kind: []string{}, MaxAttemptedBy: maxAttemptedBy, MaxToLock: maxToLock, Queue: rivercommon.QueueDefault, }) - require.NoError(t, err) require.Empty(t, jobs) jobAfter, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID}) @@ -327,13 +334,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) }) @@ -347,13 +353,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) }) @@ -369,14 +374,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) }) @@ -395,14 +399,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) }) @@ -419,13 +422,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 @@ -439,14 +441,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") }) @@ -466,13 +466,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] @@ -502,13 +501,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] @@ -535,13 +533,12 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx }) job2 := 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) // Result order isn't guaranteed by every driver. sort.Slice(jobRows, func(i, j int) bool { return jobRows[i].ID < jobRows[j].ID }) @@ -561,6 +558,48 @@ func exerciseJobRead[TTx any](ctx context.Context, t *testing.T, executorWithTx 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) { @@ -690,49 +729,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) - 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)}) + var ( + horizon = time.Now().UTC() + beforeHorizon = horizon.Add(-1 * time.Minute) + afterHorizon = horizon.Add(1 * time.Minute) + ) - t.Logf("horizon = %s", horizon) - t.Logf("stuckJob1 = %s", stuckJob1.AttemptedAt) - t.Logf("stuckJob2 = %s", stuckJob2.AttemptedAt) + 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("stuckJob1 full = %s", spew.Sdump(stuckJob1)) + t.Logf("horizon = %s", horizon) + t.Logf("stuckJob1 = %s", stuckJob1.AttemptedAt) + t.Logf("stuckJob2 = %s", stuckJob2.AttemptedAt) - // 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)}) + t.Logf("stuckJob1 full = %s", spew.Sdump(stuckJob1)) - // Not stuck because not in running state. - _ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{State: new(rivertype.JobStateAvailable)}) + // 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 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, + }) + require.NoError(t, err) + require.Equal(t, []int64{stuckJob1.ID, stuckJob2.ID}, + sliceutil.Map(stuckJobs, func(j *rivertype.JobRow) int64 { return j.ID })) - // Max two stuck - stuckJobs, err := exec.JobGetStuck(ctx, &riverdriver.JobGetStuckParams{ - Max: 2, - StuckHorizon: horizon, + 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 2512ff318..bceab7a3e 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) @@ -889,6 +891,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 +972,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 41c50b1a7..cdde6c960 100644 --- a/riverdriver/riverdrivertest/riverdrivertest.go +++ b/riverdriver/riverdrivertest/riverdrivertest.go @@ -129,3 +129,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 ab15148ed..4b8be434f 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/riverdriver/riverpgxv5/river_pgx_v5_driver.go @@ -333,7 +333,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, Kind: params.Kind, @@ -345,7 +345,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) { @@ -386,7 +386,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) { @@ -722,7 +722,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) { @@ -1335,17 +1335,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 } } @@ -1379,7 +1411,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 08669f10c..8cb51420b 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -613,7 +613,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 caf04ef2a..fb0fcad6d 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -1700,7 +1700,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 66dd461f3..90e5034e0 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -546,7 +546,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) { kindsJSON, err := json.Marshal(params.Kind) if err != nil { return nil, err @@ -572,7 +572,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) { @@ -613,7 +613,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) { @@ -1010,10 +1010,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) } @@ -1687,7 +1687,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() @@ -1697,14 +1728,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 } } @@ -1716,15 +1749,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{ @@ -1734,7 +1769,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), @@ -1746,7 +1781,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 69d48d3b1..04f667f8f 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 af2a92fe8..12cf4ee2e 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 ae7d4254e..fc1c350f9 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) } From 0ab750474f2b647fec93bb7f186679699d72ff15 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Thu, 24 Sep 2026 19:06:55 -0500 Subject: [PATCH 3/4] tolerate undecodable jobs in SQLite scheduler A job whose attempt fails because its row can't be decoded is moved to `retryable` with its values left in place. On SQLite, the scheduler decodes every job it schedules strictly, so once such a job's retry comes due, each scheduler run fails on it and rolls back the whole batch. No retryable or scheduled job gets scheduled again until the row is repaired, and a job discarded for a unique conflict fails the same way. Decode the rows the scheduler updates the same way as stuck jobs, with the fields that can't be decoded left empty instead of failing, so an undecodable job is scheduled or discarded along with the others and comes back through the fetch path to be retried or discarded. A driver test schedules undecodable retryable and scheduled jobs, and one that's discarded for a unique conflict, alongside a normal job. --- CHANGELOG.md | 2 +- riverdriver/riverdrivertest/job_update.go | 61 +++++++++++++++++++ .../riversqlite/river_sqlite_driver.go | 24 ++++---- 3 files changed, 74 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9349db6e0..ff8292768 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,7 +42,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed SQLite reusing the ID of a deleted job when that job held the largest ID, which could cause an ID observed earlier to refer to an unrelated job later. [PR #1390](https://github.com/riverqueue/river/pull/1390). - Fixed the `Job appears to be stuck` log line reporting the client-level `JobTimeout` instead of the worker-level timeout when a worker overrides `Timeout`. [PR #1394](https://github.com/riverqueue/river/pull/1394). - 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 the rescuer no longer fails on stuck jobs that can't be decoded. [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/riverdriver/riverdrivertest/job_update.go b/riverdriver/riverdrivertest/job_update.go index bceab7a3e..70e393b83 100644 --- a/riverdriver/riverdrivertest/job_update.go +++ b/riverdriver/riverdrivertest/job_update.go @@ -712,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() diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index 90e5034e0..e9bc8ba1b 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -878,6 +878,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) @@ -906,10 +915,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} } @@ -923,10 +929,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} } } @@ -938,10 +941,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} } } From 22d0b41293bd9aa7040bc810605b95d0a508d8c0 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Mon, 28 Sep 2026 11:40:19 -0500 Subject: [PATCH 4/4] append rescue errors to malformed SQLite rows A stuck SQLite job can have an `errors` JSON value that is not an array. The rescuer can read the job partially, but appending its rescue error through an array path leaves an object-valued field unchanged. That loses the rescue error even though the job changes state. Wrap a non-array value in an array before appending the rescue error, matching the failed-attempt update path. Exercise a mixed rescue batch through both SQLite drivers and verify the existing value and new error remain readable. --- CHANGELOG.md | 2 +- riverdriver/riverdrivertest/job_update.go | 38 +++++++++++++++++++ .../riversqlite/internal/dbsqlc/river_job.sql | 4 +- .../internal/dbsqlc/river_job.sql.go | 4 +- 4 files changed, 45 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff8292768..d47032e90 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,7 +42,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed SQLite reusing the ID of a deleted job when that job held the largest ID, which could cause an ID observed earlier to refer to an unrelated job later. [PR #1390](https://github.com/riverqueue/river/pull/1390). - Fixed the `Job appears to be stuck` log line reporting the client-level `JobTimeout` instead of the worker-level timeout when a worker overrides `Timeout`. [PR #1394](https://github.com/riverqueue/river/pull/1394). - 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). +- 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 original values are preserved, with non-array `errors` values wrapped to append the failure or rescue error. Neither the rescuer nor the SQLite scheduler fails on such a job, so its 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/riverdriver/riverdrivertest/job_update.go b/riverdriver/riverdrivertest/job_update.go index 70e393b83..dbc455d10 100644 --- a/riverdriver/riverdrivertest/job_update.go +++ b/riverdriver/riverdrivertest/job_update.go @@ -319,6 +319,44 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT } }) + // SQLite can hold a non-array errors value on a stuck job. Rescue must + // preserve it and append the rescue error while updating the whole batch. + t.Run("JobRescueMany_NonArrayErrorsWrapped", 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 + attemptedAt := now.Add(-time.Hour) + badJob := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &attemptedAt, State: new(rivertype.JobStateRunning)}) + goodJob := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{AttemptedAt: &attemptedAt, State: new(rivertype.JobStateRunning)}) + sqliteSetJobJSONColumn(ctx, t, exec, badJob.ID, "errors", `{"error":"existing value"}`) + + _, err := exec.JobRescueMany(ctx, &riverdriver.JobRescueManyParams{ + ID: []int64{badJob.ID, goodJob.ID}, + Error: [][]byte{[]byte(`{"error":"rescue error"}`), []byte(`{"error":"rescue error"}`)}, + FinalizedAt: []*time.Time{nil, nil}, + ScheduledAt: []time.Time{now, now}, + State: []string{string(rivertype.JobStateRetryable), string(rivertype.JobStateRetryable)}, + StuckHorizon: now, + }) + require.NoError(t, err) + + badAfter, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: badJob.ID}) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateRetryable, badAfter.State) + require.Equal(t, []string{"existing value", "rescue error"}, + sliceutil.Map(badAfter.Errors, func(e rivertype.AttemptError) string { return e.Error })) + + goodAfter, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: goodJob.ID}) + require.NoError(t, err) + require.Equal(t, rivertype.JobStateRetryable, goodAfter.State) + require.Equal(t, "rescue error", goodAfter.Errors[0].Error) + }) + t.Run("JobRescueMany_ReclaimedAfterFetch", func(t *testing.T) { t.Parallel() diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index 8cb51420b..21fcbb1e8 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -498,7 +498,9 @@ LIMIT @max; -- name: JobRescue :exec UPDATE /* TEMPLATE: schema */river_job SET - errors = jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(@error))), + errors = CASE WHEN coalesce(json_type(errors), 'array') <> 'array' + THEN jsonb(json_array(json(errors), json(@error))) + ELSE jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(@error))) END, finalized_at = cast(sqlc.narg('finalized_at') as text), scheduled_at = cast(@scheduled_at AS text), metadata = jsonb_set( diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go index fb0fcad6d..74d9b793c 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -1300,7 +1300,9 @@ func (q *Queries) JobList(ctx context.Context, db DBTX, max int64) ([]*RiverJob, const jobRescue = `-- name: JobRescue :exec UPDATE /* TEMPLATE: schema */river_job SET - errors = jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(?1))), + errors = CASE WHEN coalesce(json_type(errors), 'array') <> 'array' + THEN jsonb(json_array(json(errors), json(?1))) + ELSE jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(?1))) END, finalized_at = cast(?2 as text), scheduled_at = cast(?3 AS text), metadata = jsonb_set(