allow customizing processing order

client.go

@@ -361,6 +361,10 @@ type Config struct {
 	// Scheduler run interval. Shared between the scheduler and producer/job
 	// executors, but not currently exposed for configuration.
 	schedulerInterval time.Duration
+
+	// Customize processing order of job. Jobs can be processed in a FIFO/LIFO
+	// fashion
+	Fifo bool
 }
 
 // WithDefaults returns a copy of the Config with all default values applied.
@@ -2032,6 +2036,7 @@ func (c *Client[TTx]) addProducer(queueName string, queueConfig QueueConfig) (*p
 		Schema:                       c.config.Schema,
 		StaleProducerRetentionPeriod: 5 * time.Minute,
 		Workers:                      c.config.Workers,
+		Fifo:                         c.config.Fifo,
 	})
 	c.producersByQueueName[queueName] = producer
 	return producer, nil

producer.go

@@ -112,6 +112,8 @@ type producerConfig struct {
 	Schema                       string
 	StaleProducerRetentionPeriod time.Duration
 	Workers                      *Workers
+
+	Fifo bool
 }
 
 func (c *producerConfig) mustValidate() *producerConfig {
@@ -754,6 +756,7 @@ func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultC
 		Queue:          p.config.Queue,
 		ProducerID:     p.id.Load(),
 		Schema:         p.config.Schema,
+		Fifo:           p.config.Fifo,
 	})
 	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

@@ -372,6 +372,7 @@ type JobGetAvailableParams struct {
 	ProducerID     int64
 	Queue          string
 	Schema         string
+	Fifo           bool
 }
 
 type JobGetByIDParams struct {

riverdriver/riverdatabasesql/river_database_sql_driver.go

@@ -240,17 +240,31 @@ 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,
-		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)
+	if params.Fifo {
+		jobs, err := dbsqlc.New().JobGetAvailableFifo(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetAvailableFifoParams{
+			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)
+		}
+		return sliceutil.MapError(jobs, jobRowFromInternal)
+	} else {
+		jobs, err := dbsqlc.New().JobGetAvailableLifo(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetAvailableLifoParams{
+			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)
+		}
+		return sliceutil.MapError(jobs, jobRowFromInternal)
 	}
-	return sliceutil.MapError(jobs, jobRowFromInternal)
 }
 
 func (e *Executor) JobGetByID(ctx context.Context, params *riverdriver.JobGetByIDParams) (*rivertype.JobRow, error) {

riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql

@@ -145,7 +145,7 @@ FROM /* TEMPLATE: schema */river_job
 WHERE id IN (SELECT id FROM deleted_jobs)
 ORDER BY /* TEMPLATE_BEGIN: order_by_clause */ id /* TEMPLATE_END */;
 
--- name: JobGetAvailable :many
+-- name: JobGetAvailableLifo :many
 WITH locked_jobs AS (
     SELECT
         *
@@ -184,6 +184,45 @@ WHERE
 RETURNING
     river_job.*;
 
+-- name: JobGetAvailableFifo :many
+WITH locked_jobs AS (
+    SELECT
+        *
+    FROM
+        /* TEMPLATE: schema */river_job
+    WHERE
+        state = 'available'
+        AND queue = @queue::text
+        AND scheduled_at <= coalesce(sqlc.narg('now')::timestamptz, now())
+    ORDER BY
+        priority ASC,
+        scheduled_at DESC,
+        id ASC
+    LIMIT @max_to_lock::integer
+    FOR UPDATE
+    SKIP LOCKED
+)
+UPDATE
+    /* TEMPLATE: schema */river_job
+SET
+    state = 'running',
+    attempt = river_job.attempt + 1,
+    attempted_at = coalesce(sqlc.narg('now')::timestamptz, now()),
+    attempted_by = array_append(
+        CASE WHEN array_length(river_job.attempted_by, 1) >= @max_attempted_by::int
+        -- +2 instead of +1 because Postgres array indexing starts at 1, not 0.
+        THEN river_job.attempted_by[array_length(river_job.attempted_by, 1) + 2 - @max_attempted_by:]
+        ELSE river_job.attempted_by
+        END,
+        @attempted_by::text
+    )
+FROM
+    locked_jobs
+WHERE
+    river_job.id = locked_jobs.id
+RETURNING
+    river_job.*;
+
 -- name: JobGetByID :one
 SELECT *
 FROM /* TEMPLATE: schema */river_job