Add maximum bound for attempted_by array

As described by #972, it may be possible for huge numbers of snoozes to
blow out a job row's `attempted_by` array as a job is locked over and
over again. Multiplied by many jobs, this can produce vast quantities of
data that gets sent over the network.

Here, put in a ratchet on `attempted_by` so that if the array becomes
larger than 100 elements, we knock the oldest one off in favor of the
most recent client and the most fresh 99.

Unfortunately the implementation isn't particularly clean in either
Postgres or SQLite. In Postgres it would've been cleaner if we'd had the
`attempted_by` in reverse order so the new client was on front because
the built-in array functions would be friendlier to that layout, but
because it's not, we have to do something a little hackier involving a
`CASE` statement instead.

SQLite is even worse. SQLite has no array functions at all, which
doesn't help, but moreover every strategy I tried ended up blocked by a
sqlc SQLite bug, so after trying everything I could think of, I ended up
having to extract the piece that does the array truncation into a SQL
template string to get this over the line. This could be removed in the
future if any one of a number of outstanding sqlc bugs are fixed (e.g.
[1]).

[1] sqlc-dev/sqlc#3610

Author: brandur

.golangci.yaml

@@ -44,6 +44,9 @@ linters:
             - desc: Don't use `riverinternaltest` package outside of test environments.
               pkg: github.com/riverqueue/river/internal/riverinternaltest
 
+    exhaustive:
+      default-signifies-exhaustive: true
+
     forbidigo:
       forbid:
         - msg: Use `require` variants instead.

CHANGELOG.md

@@ -23,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - The reindexer maintenance service now reindexes all `river_job` indexes, including its primary key. This is expected to help in situations where the jobs table has in the past expanded to a very large size (which makes most indexes larger), is now a much more modest size, but has left the indexes in their expanded state. [PR #963](https://github.com/riverqueue/river/pull/963).
 - The River CLI now accepts a `--target-version` of 0 with `river migrate-down` to run all down migrations and remove all River tables (previously, -1 was used for this; -1 still works, but now 0 also works). [PR #966](https://github.com/riverqueue/river/pull/966).
 - **Breaking change:** The `HookWorkEnd` interface's `WorkEnd` function now receives a `JobRow` parameter in addition to the `error` it received before. Having a `JobRow` to work with is fairly crucial to most functionality that a hook would implement, and its previous omission was entirely an error. [PR #970](https://github.com/riverqueue/river/pull/970).
+- Add maximum bound to each job's `attempted_by` array so that in degenerate cases where a job is locked many, many times (say it's snoozed hundreds of times), it doesn't grow to unlimited bounds. [PR #974](https://github.com/riverqueue/river/pull/974).
 
 ### Fixed
 

internal/jobexecutor/job_executor_test.go

@@ -159,9 +159,9 @@ func TestJobExecutor_Execute(t *testing.T) {
 
 		// Fetch the job to make sure it's marked as running:
 		jobs, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{
-			Max:   1,
-			Now:   ptrutil.Ptr(now),
-			Queue: rivercommon.QueueDefault,
+			MaxToLock: 1,
+			Now:       ptrutil.Ptr(now),
+			Queue:     rivercommon.QueueDefault,
 		})
 		require.NoError(t, err)
 

internal/maintenance/job_scheduler_test.go

@@ -329,9 +329,10 @@ func TestJobScheduler(t *testing.T) {
 		addJob := func(queue string, fromNow time.Duration, state rivertype.JobState) {
 			t.Helper()
 			var finalizedAt *time.Time
-			switch state { //nolint:exhaustive
+			switch state {
 			case rivertype.JobStateCompleted, rivertype.JobStateCancelled, rivertype.JobStateDiscarded:
 				finalizedAt = ptrutil.Ptr(now.Add(fromNow))
+			default:
 			}
 			testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
 				FinalizedAt: finalizedAt,

job_list_params.go

@@ -234,7 +234,6 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) {
 	if p.sortField == JobListOrderByFinalizedAt {
 		currentNonFinalizedStates := make([]rivertype.JobState, 0, len(p.states))
 		for _, state := range p.states {
-			//nolint:exhaustive
 			switch state {
 			case rivertype.JobStateCancelled, rivertype.JobStateCompleted, rivertype.JobStateDiscarded:
 			default:

producer.go

@@ -742,13 +742,18 @@ func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultC
 	// back to the queue.
 	ctx := context.WithoutCancel(workCtx)
 
+	// Maximum size of the `attempted_by` array on each job row. This maximum is
+	// rarely hit, but exists to protect against degenerate cases.
+	const maxAttemptedBy = 100
+
 	jobs, err := p.pilot.JobGetAvailable(ctx, p.exec, p.state, &riverdriver.JobGetAvailableParams{
-		ClientID:   p.config.ClientID,
-		Max:        count,
-		Now:        p.Time.NowUTCOrNil(),
-		Queue:      p.config.Queue,
-		ProducerID: p.id.Load(),
-		Schema:     p.config.Schema,
+		ClientID:       p.config.ClientID,
+		MaxAttemptedBy: maxAttemptedBy,
+		MaxToLock:      count,
+		Now:            p.Time.NowUTCOrNil(),
+		Queue:          p.config.Queue,
+		ProducerID:     p.id.Load(),
+		Schema:         p.config.Schema,
 	})
 	if err != nil {
 		p.Logger.Error(p.Name+": Error fetching jobs", slog.String("err", err.Error()), slog.String("queue", p.config.Queue))

riverdriver/river_driver_interface.go

@@ -365,12 +365,13 @@ type JobDeleteManyParams struct {
 }
 
 type JobGetAvailableParams struct {
-	ClientID   string
-	Max        int
-	Now        *time.Time
-	ProducerID int64
-	Queue      string
-	Schema     string
+	ClientID       string
+	MaxAttemptedBy int
+	MaxToLock      int
+	Now            *time.Time
+	ProducerID     int64
+	Queue          string
+	Schema         string
 }
 
 type JobGetByIDParams struct {

riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go

@@ -241,10 +241,11 @@ func (e *Executor) JobDeleteMany(ctx context.Context, params *riverdriver.JobDel
 
 func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) {
 	jobs, err := dbsqlc.New().JobGetAvailable(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetAvailableParams{
-		AttemptedBy: params.ClientID,
-		Max:         int32(min(params.Max, math.MaxInt32)), //nolint:gosec
-		Now:         params.Now,
-		Queue:       params.Queue,
+		AttemptedBy:    params.ClientID,
+		MaxAttemptedBy: int32(min(params.MaxAttemptedBy, math.MaxInt32)), //nolint:gosec
+		MaxToLock:      int32(min(params.MaxToLock, math.MaxInt32)),      //nolint:gosec
+		Now:            params.Now,
+		Queue:          params.Queue,
 	})
 	if err != nil {
 		return nil, interpretError(err)

riverdriver/riverdatabasesql/river_database_sql_driver.go

@@ -399,9 +399,9 @@ func BenchmarkDriverRiverPgxV5_Executor(b *testing.B) {
 
 		for range b.N {
 			if _, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{
-				ClientID: clientID,
-				Max:      100,
-				Queue:    river.QueueDefault,
+				ClientID:  clientID,
+				MaxToLock: 100,
+				Queue:     river.QueueDefault,
 			}); err != nil {
 				b.Fatal(err)
 			}
@@ -425,9 +425,9 @@ func BenchmarkDriverRiverPgxV5_Executor(b *testing.B) {
 		b.RunParallel(func(pb *testing.PB) {
 			for pb.Next() {
 				if _, err := exec.JobGetAvailable(ctx, &riverdriver.JobGetAvailableParams{
-					ClientID: clientID,
-					Max:      100,
-					Queue:    river.QueueDefault,
+					ClientID:  clientID,
+					MaxToLock: 100,
+					Queue:     river.QueueDefault,
 				}); err != nil {
 					b.Fatal(err)
 				}