diff --git a/client.go b/client.go index be3dc0f5..883d32a9 100644 --- a/client.go +++ b/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 diff --git a/producer.go b/producer.go index 0bd1d680..db8f316c 100644 --- a/producer.go +++ b/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)) diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index d2ae1e44..d24b674d 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -372,6 +372,7 @@ type JobGetAvailableParams struct { ProducerID int64 Queue string Schema string + Fifo bool } type JobGetByIDParams struct { diff --git a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go index 86c70cc2..90526dab 100644 --- a/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go @@ -257,7 +257,103 @@ func (q *Queries) JobDeleteMany(ctx context.Context, db DBTX, max int32) ([]*Riv return items, nil } -const jobGetAvailable = `-- name: JobGetAvailable :many +const jobGetAvailableFifo = `-- name: JobGetAvailableFifo :many +WITH locked_jobs AS ( + SELECT + id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states + FROM + /* TEMPLATE: schema */river_job + WHERE + state = 'available' + AND queue = $4::text + AND scheduled_at <= coalesce($1::timestamptz, now()) + ORDER BY + priority ASC, + scheduled_at DESC, + id ASC + LIMIT $5::integer + FOR UPDATE + SKIP LOCKED +) +UPDATE + /* TEMPLATE: schema */river_job +SET + state = 'running', + attempt = river_job.attempt + 1, + attempted_at = coalesce($1::timestamptz, now()), + attempted_by = array_append( + CASE WHEN array_length(river_job.attempted_by, 1) >= $2::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 - $2:] + ELSE river_job.attempted_by + END, + $3::text + ) +FROM + locked_jobs +WHERE + river_job.id = locked_jobs.id +RETURNING + river_job.id, river_job.args, river_job.attempt, river_job.attempted_at, river_job.attempted_by, river_job.created_at, river_job.errors, river_job.finalized_at, river_job.kind, river_job.max_attempts, river_job.metadata, river_job.priority, river_job.queue, river_job.state, river_job.scheduled_at, river_job.tags, river_job.unique_key, river_job.unique_states +` + +type JobGetAvailableFifoParams struct { + Now *time.Time + MaxAttemptedBy int32 + AttemptedBy string + Queue string + MaxToLock int32 +} + +func (q *Queries) JobGetAvailableFifo(ctx context.Context, db DBTX, arg *JobGetAvailableFifoParams) ([]*RiverJob, error) { + rows, err := db.QueryContext(ctx, jobGetAvailableFifo, + arg.Now, + arg.MaxAttemptedBy, + arg.AttemptedBy, + arg.Queue, + arg.MaxToLock, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []*RiverJob + for rows.Next() { + var i RiverJob + if err := rows.Scan( + &i.ID, + &i.Args, + &i.Attempt, + &i.AttemptedAt, + pq.Array(&i.AttemptedBy), + &i.CreatedAt, + pq.Array(&i.Errors), + &i.FinalizedAt, + &i.Kind, + &i.MaxAttempts, + &i.Metadata, + &i.Priority, + &i.Queue, + &i.State, + &i.ScheduledAt, + pq.Array(&i.Tags), + &i.UniqueKey, + &i.UniqueStates, + ); err != nil { + return nil, err + } + items = append(items, &i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const jobGetAvailableLifo = `-- name: JobGetAvailableLifo :many WITH locked_jobs AS ( SELECT id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states @@ -297,7 +393,7 @@ RETURNING river_job.id, river_job.args, river_job.attempt, river_job.attempted_at, river_job.attempted_by, river_job.created_at, river_job.errors, river_job.finalized_at, river_job.kind, river_job.max_attempts, river_job.metadata, river_job.priority, river_job.queue, river_job.state, river_job.scheduled_at, river_job.tags, river_job.unique_key, river_job.unique_states ` -type JobGetAvailableParams struct { +type JobGetAvailableLifoParams struct { Now *time.Time MaxAttemptedBy int32 AttemptedBy string @@ -305,8 +401,8 @@ type JobGetAvailableParams struct { MaxToLock int32 } -func (q *Queries) JobGetAvailable(ctx context.Context, db DBTX, arg *JobGetAvailableParams) ([]*RiverJob, error) { - rows, err := db.QueryContext(ctx, jobGetAvailable, +func (q *Queries) JobGetAvailableLifo(ctx context.Context, db DBTX, arg *JobGetAvailableLifoParams) ([]*RiverJob, error) { + rows, err := db.QueryContext(ctx, jobGetAvailableLifo, arg.Now, arg.MaxAttemptedBy, arg.AttemptedBy, diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go index 3620a324..929d0a71 100644 --- a/riverdriver/riverdatabasesql/river_database_sql_driver.go +++ b/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) { diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql index 61a64bbd..dc945fdc 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql +++ b/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 diff --git a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go index d451d5a8..97d74fb8 100644 --- a/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql.go @@ -254,7 +254,100 @@ func (q *Queries) JobDeleteMany(ctx context.Context, db DBTX, max int32) ([]*Riv return items, nil } -const jobGetAvailable = `-- name: JobGetAvailable :many +const jobGetAvailableFifo = `-- name: JobGetAvailableFifo :many +WITH locked_jobs AS ( + SELECT + id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states + FROM + /* TEMPLATE: schema */river_job + WHERE + state = 'available' + AND queue = $4::text + AND scheduled_at <= coalesce($1::timestamptz, now()) + ORDER BY + priority ASC, + scheduled_at DESC, + id ASC + LIMIT $5::integer + FOR UPDATE + SKIP LOCKED +) +UPDATE + /* TEMPLATE: schema */river_job +SET + state = 'running', + attempt = river_job.attempt + 1, + attempted_at = coalesce($1::timestamptz, now()), + attempted_by = array_append( + CASE WHEN array_length(river_job.attempted_by, 1) >= $2::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 - $2:] + ELSE river_job.attempted_by + END, + $3::text + ) +FROM + locked_jobs +WHERE + river_job.id = locked_jobs.id +RETURNING + river_job.id, river_job.args, river_job.attempt, river_job.attempted_at, river_job.attempted_by, river_job.created_at, river_job.errors, river_job.finalized_at, river_job.kind, river_job.max_attempts, river_job.metadata, river_job.priority, river_job.queue, river_job.state, river_job.scheduled_at, river_job.tags, river_job.unique_key, river_job.unique_states +` + +type JobGetAvailableFifoParams struct { + Now *time.Time + MaxAttemptedBy int32 + AttemptedBy string + Queue string + MaxToLock int32 +} + +func (q *Queries) JobGetAvailableFifo(ctx context.Context, db DBTX, arg *JobGetAvailableFifoParams) ([]*RiverJob, error) { + rows, err := db.Query(ctx, jobGetAvailableFifo, + arg.Now, + arg.MaxAttemptedBy, + arg.AttemptedBy, + arg.Queue, + arg.MaxToLock, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []*RiverJob + for rows.Next() { + var i RiverJob + if err := rows.Scan( + &i.ID, + &i.Args, + &i.Attempt, + &i.AttemptedAt, + &i.AttemptedBy, + &i.CreatedAt, + &i.Errors, + &i.FinalizedAt, + &i.Kind, + &i.MaxAttempts, + &i.Metadata, + &i.Priority, + &i.Queue, + &i.State, + &i.ScheduledAt, + &i.Tags, + &i.UniqueKey, + &i.UniqueStates, + ); err != nil { + return nil, err + } + items = append(items, &i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const jobGetAvailableLifo = `-- name: JobGetAvailableLifo :many WITH locked_jobs AS ( SELECT id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states @@ -294,7 +387,7 @@ RETURNING river_job.id, river_job.args, river_job.attempt, river_job.attempted_at, river_job.attempted_by, river_job.created_at, river_job.errors, river_job.finalized_at, river_job.kind, river_job.max_attempts, river_job.metadata, river_job.priority, river_job.queue, river_job.state, river_job.scheduled_at, river_job.tags, river_job.unique_key, river_job.unique_states ` -type JobGetAvailableParams struct { +type JobGetAvailableLifoParams struct { Now *time.Time MaxAttemptedBy int32 AttemptedBy string @@ -302,8 +395,8 @@ type JobGetAvailableParams struct { MaxToLock int32 } -func (q *Queries) JobGetAvailable(ctx context.Context, db DBTX, arg *JobGetAvailableParams) ([]*RiverJob, error) { - rows, err := db.Query(ctx, jobGetAvailable, +func (q *Queries) JobGetAvailableLifo(ctx context.Context, db DBTX, arg *JobGetAvailableLifoParams) ([]*RiverJob, error) { + rows, err := db.Query(ctx, jobGetAvailableLifo, arg.Now, arg.MaxAttemptedBy, arg.AttemptedBy, diff --git a/riverdriver/riverpgxv5/river_pgx_v5_driver.go b/riverdriver/riverpgxv5/river_pgx_v5_driver.go index d5620ac3..16bc223a 100644 --- a/riverdriver/riverpgxv5/river_pgx_v5_driver.go +++ b/riverdriver/riverpgxv5/river_pgx_v5_driver.go @@ -244,17 +244,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) { diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index d466830c..170ddf7a 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -95,7 +95,7 @@ RETURNING *; -- Differs from the Postgres version in that we don't have `FOR UPDATE SKIP -- LOCKED`. It doesn't exist in SQLite, but more aptly, there's only one writer -- on SQLite at a time, so nothing else has the rows locked. --- name: JobGetAvailable :many +-- name: JobGetAvailableLifo :many UPDATE /* TEMPLATE: schema */river_job SET attempt = river_job.attempt + 1, @@ -122,6 +122,36 @@ WHERE id IN ( ) RETURNING *; +-- Differs from the Postgres version in that we don't have `FOR UPDATE SKIP +-- LOCKED`. It doesn't exist in SQLite, but more aptly, there's only one writer +-- on SQLite at a time, so nothing else has the rows locked. +-- name: JobGetAvailableFifo :many +UPDATE /* TEMPLATE: schema */river_job +SET + attempt = river_job.attempt + 1, + attempted_at = coalesce(cast(sqlc.narg('now') AS text), datetime('now', 'subsec')), + + -- This is replaced in the driver to work around sqlc bugs for SQLite. See + -- comments there for more details. + attempted_by = /* TEMPLATE_BEGIN: attempted_by_clause */ attempted_by /* TEMPLATE_END */, + + state = 'running' +WHERE id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_job + WHERE + priority >= 0 + AND river_job.queue = @queue + AND scheduled_at <= coalesce(cast(sqlc.narg('now') AS text), datetime('now', 'subsec')) + AND state = 'available' + ORDER BY + priority ASC, + scheduled_at DESC, + id ASC + LIMIT @max_to_lock +) +RETURNING *; + -- name: JobGetByID :one SELECT * FROM /* TEMPLATE: schema */river_job diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go index cdea9049..e06ac54b 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -205,7 +205,86 @@ func (q *Queries) JobDeleteMany(ctx context.Context, db DBTX, max int64) ([]*Riv return items, nil } -const jobGetAvailable = `-- name: JobGetAvailable :many +const jobGetAvailableFifo = `-- name: JobGetAvailableFifo :many +UPDATE /* TEMPLATE: schema */river_job +SET + attempt = river_job.attempt + 1, + attempted_at = coalesce(cast(?1 AS text), datetime('now', 'subsec')), + + -- This is replaced in the driver to work around sqlc bugs for SQLite. See + -- comments there for more details. + attempted_by = /* TEMPLATE_BEGIN: attempted_by_clause */ attempted_by /* TEMPLATE_END */, + + state = 'running' +WHERE id IN ( + SELECT id + FROM /* TEMPLATE: schema */river_job + WHERE + priority >= 0 + AND river_job.queue = ?2 + AND scheduled_at <= coalesce(cast(?1 AS text), datetime('now', 'subsec')) + AND state = 'available' + ORDER BY + priority ASC, + scheduled_at DESC, + id ASC + LIMIT ?3 +) +RETURNING id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states +` + +type JobGetAvailableFifoParams struct { + Now *string + Queue string + MaxToLock int64 +} + +// Differs from the Postgres version in that we don't have `FOR UPDATE SKIP +// LOCKED`. It doesn't exist in SQLite, but more aptly, there's only one writer +// on SQLite at a time, so nothing else has the rows locked. +func (q *Queries) JobGetAvailableFifo(ctx context.Context, db DBTX, arg *JobGetAvailableFifoParams) ([]*RiverJob, error) { + rows, err := db.QueryContext(ctx, jobGetAvailableFifo, arg.Now, arg.Queue, arg.MaxToLock) + if err != nil { + return nil, err + } + defer rows.Close() + var items []*RiverJob + for rows.Next() { + var i RiverJob + if err := rows.Scan( + &i.ID, + &i.Args, + &i.Attempt, + &i.AttemptedAt, + &i.AttemptedBy, + &i.CreatedAt, + &i.Errors, + &i.FinalizedAt, + &i.Kind, + &i.MaxAttempts, + &i.Metadata, + &i.Priority, + &i.Queue, + &i.State, + &i.ScheduledAt, + &i.Tags, + &i.UniqueKey, + &i.UniqueStates, + ); err != nil { + return nil, err + } + items = append(items, &i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const jobGetAvailableLifo = `-- name: JobGetAvailableLifo :many UPDATE /* TEMPLATE: schema */river_job SET attempt = river_job.attempt + 1, @@ -233,7 +312,7 @@ WHERE id IN ( RETURNING id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states ` -type JobGetAvailableParams struct { +type JobGetAvailableLifoParams struct { Now *string Queue string MaxToLock int64 @@ -242,8 +321,8 @@ type JobGetAvailableParams struct { // Differs from the Postgres version in that we don't have `FOR UPDATE SKIP // LOCKED`. It doesn't exist in SQLite, but more aptly, there's only one writer // on SQLite at a time, so nothing else has the rows locked. -func (q *Queries) JobGetAvailable(ctx context.Context, db DBTX, arg *JobGetAvailableParams) ([]*RiverJob, error) { - rows, err := db.QueryContext(ctx, jobGetAvailable, arg.Now, arg.Queue, arg.MaxToLock) +func (q *Queries) JobGetAvailableLifo(ctx context.Context, db DBTX, arg *JobGetAvailableLifoParams) ([]*RiverJob, error) { + rows, err := db.QueryContext(ctx, jobGetAvailableLifo, arg.Now, arg.Queue, arg.MaxToLock) if err != nil { return nil, err } diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index e093761e..97d358b0 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -382,16 +382,27 @@ func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobG "attempted_by": params.ClientID, "max_attempted_by": params.MaxAttemptedBy, }) - - jobs, err := dbsqlc.New().JobGetAvailable(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobGetAvailableParams{ - MaxToLock: int64(params.MaxToLock), - Now: timeStringNullable(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{ + MaxToLock: int64(params.MaxToLock), + Now: timeStringNullable(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{ + MaxToLock: int64(params.MaxToLock), + Now: timeStringNullable(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) {