From 9874f198eb82647e5d3153dea6d2b9ea05bda90d Mon Sep 17 00:00:00 2001 From: Brandur Date: Fri, 19 Apr 2024 18:09:14 -0700 Subject: [PATCH] Add capacity to list jobs by ID + make default Here, in addition to the new job list sort orders added by #304, add one more for sorting by job ID, and make it the default. This is another breaking change in the job list API, and I wouldn't normally make it at this point, but our next release will contain a number of breaking changes, including to the job list API, so the time is now. My rationale for the change: * Listing and paginating by ID is an overwhelmingly common pattern in web and database APIs, and I think it being the default would be more intuitive for more people. * Ordering by ID is more ergonomic because no `JobListParams.States` invocation needs to made. It's shorter, and especially when a fairly normal use case will be to iterate across all job rows, this is a minor nicety. * Ordering by ID allows the entire job collection to be iterated regardless of job state. `JobListOrderByTime` requires a state, and some order is needed for cursors to work. * The behavior of `JobListOrderByTime` is a little odd in that it changes dynamically based on the requested list states, and there's no way to intuit what the order will be without knowing a lot about River internals and thinking very carefully about it. Furthermore, the time that'd be chosen wasn't documented anywhere, so the only way to know for sure what it would be was to read River's source code. * With the inclusion of #304, `JobListOrderByTime`'s behavior has gotten even a little more surprising because the state to be chosen to select a timestamp was the _first_ value sent to `JobListParams.States`, with any additional values sent ignored, also creating a somewhat nonsensical result (e.g. `States(running, available)` would select `attempted_at`, but would be `NULL` for any jobs in the `available` state). This behavior was not documented. I also found a bug that was a hold over from #304 as more than one sort order became available. The function `JobListCursorFromJob` took a sort order, but would produce the wrong result unless the user remembered to set the exact same sort order on their job list parameters. For example, this would do the wrong thing: res, err = client.JobList(ctx, NewJobListParams().After(JobListCursorFromJob(job4, JobListOrderByScheduledAt))) `JobListCursorFromJob` would extract `scheduled_at` from `job4`, but then list using to the default job list order. Previously that was based on time, so this result would've been wrong _unless_ the job list parameters filtered to state `available`, `retryable`, or `scheduled` so that `scheduled_at` was also used when comparing to other jobs. A caller could compensate by specifying sort order in both places, but this is pretty ugly, and there was no check to make sure that the list paramaters and cursor shared the same order: res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderAsc).After(JobListCursorFromJob(job4, JobListOrderByScheduledAt))) The corrected version of this doesn't use an order when initializing the cursor, instead using the one from the job list params, meaning that the same time field is always used between list query and what's extracted from the cursor's job row. res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderAsc).After(JobListCursorFromJob(job4))) This also makes the invocation shorter and more ergonomic to call. Along with the above, we also make a few tweaks to documentation. `JobListOrderByTime` now documents which timestamps it uses, and indicates that only the first value sent to `JobListParams.States` will be respected. --- CHANGELOG.md | 5 +- client.go | 4 +- client_test.go | 109 ++++++++++++++++++++++++++++++------ job_list_params.go | 120 +++++++++++++++++++++++++++++----------- job_list_params_test.go | 74 +++++++++++++++++++++---- 5 files changed, 248 insertions(+), 64 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 36d9d54d..0dd89c0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- **Breaking change:** JobList/JobListTx now support querying Jobs by a list of Job Kinds and States. Also allows for filtering by specific timestamp values. Thank you Jos Kraaijeveld (@thatjos)! 🙏🏻 [PR #236](https://github.com/riverqueue/river/pull/236). +- **Breaking change:** There are a number of small breaking changes in the job list API using `JobList`/`JobListTx`: + - Now support querying jobs by a list of Job Kinds and States. Also allows for filtering by specific timestamp values. Thank you Jos Kraaijeveld (@thatjos)! 🙏🏻 [PR #236](https://github.com/riverqueue/river/pull/236). + - Job listing now defaults to ordering by job ID (`JobListOrderByID`) instead of a job timestamp dependent on on requested job state. The previous ordering behavior is still available with `NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)`. [PR #XXX](https://github.com/riverqueue/river/pull/XXX). + - The function `JobListCursorFromJob` no longer needs a sort order parameter. Instead, sort order is determined based on the job list parameters that the cursor is subsequently used with. [PR #XXX](https://github.com/riverqueue/river/pull/XXX). - **Breaking change:** Client `Insert` and `InsertTx` functions now return a `JobInsertResult` struct instead of a `JobRow`. This allows the result to include metadata like the new `UniqueSkippedAsDuplicate` property, so callers can tell whether an inserted job was skipped due to unique constraint. [PR #292](https://github.com/riverqueue/river/pull/292). - **Breaking change:** Client `InsertMany` and `InsertManyTx` now return number of jobs inserted as `int` instead of `int64`. This change was made to make the type in use a little more idiomatic. [PR #293](https://github.com/riverqueue/river/pull/293). - **Breaking change:** `river.JobState*` type aliases have been removed. All job state constants should be accessed through `rivertype.JobState*` instead. [PR #300](https://github.com/riverqueue/river/pull/300). diff --git a/client.go b/client.go index acebd845..6382ac17 100644 --- a/client.go +++ b/client.go @@ -1489,7 +1489,7 @@ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) (*JobL } res := &JobListResult{Jobs: jobs} if len(jobs) > 0 { - res.LastCursor = JobListCursorFromJob(jobs[len(jobs)-1], params.sortField) + res.LastCursor = jobListCursorFromJobAndParams(jobs[len(jobs)-1], params) } return res, nil } @@ -1519,7 +1519,7 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara } res := &JobListResult{Jobs: jobs} if len(jobs) > 0 { - res.LastCursor = JobListCursorFromJob(jobs[len(jobs)-1], params.sortField) + res.LastCursor = jobListCursorFromJobAndParams(jobs[len(jobs)-1], params) } return res, nil } diff --git a/client_test.go b/client_test.go index c9ddafa4..708866af 100644 --- a/client_test.go +++ b/client_test.go @@ -1500,7 +1500,24 @@ func Test_Client_JobList(t *testing.T) { require.Equal(t, []int64{job3.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) }) - t.Run("SortsAvailableRetryableAndScheduledJobsByScheduledAt", func(t *testing.T) { + t.Run("DefaultsToOrderingByID", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) + require.NoError(t, err) + require.Equal(t, []int64{job1.ID, job2.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderDesc)) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + }) + + t.Run("OrderByTimeSortsAvailableRetryableAndScheduledJobsByScheduledAt", func(t *testing.T) { t.Parallel() client, bundle := setup(t) @@ -1516,7 +1533,7 @@ func Test_Client_JobList(t *testing.T) { job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(state), ScheduledAt: &now}) job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(state), ScheduledAt: ptrutil.Ptr(now.Add(-5 * time.Second))}) - res, err := client.JobList(ctx, NewJobListParams().States(state)) + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(state)) require.NoError(t, err) require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) @@ -1526,7 +1543,7 @@ func Test_Client_JobList(t *testing.T) { } }) - t.Run("SortsCancelledCompletedAndDiscardedJobsByFinalizedAt", func(t *testing.T) { + t.Run("OrderByTimeSortsCancelledCompletedAndDiscardedJobsByFinalizedAt", func(t *testing.T) { t.Parallel() client, bundle := setup(t) @@ -1542,7 +1559,7 @@ func Test_Client_JobList(t *testing.T) { job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(state), FinalizedAt: ptrutil.Ptr(now.Add(-10 * time.Second))}) job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(state), FinalizedAt: ptrutil.Ptr(now.Add(-15 * time.Second))}) - res, err := client.JobList(ctx, NewJobListParams().States(state)) + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(state)) require.NoError(t, err) require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) @@ -1552,7 +1569,7 @@ func Test_Client_JobList(t *testing.T) { } }) - t.Run("SortsRunningJobsByAttemptedAt", func(t *testing.T) { + t.Run("OrderByTimeSortsRunningJobsByAttemptedAt", func(t *testing.T) { t.Parallel() client, bundle := setup(t) @@ -1561,7 +1578,7 @@ func Test_Client_JobList(t *testing.T) { job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRunning), AttemptedAt: &now}) job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRunning), AttemptedAt: ptrutil.Ptr(now.Add(-5 * time.Second))}) - res, err := client.JobList(ctx, NewJobListParams().States(rivertype.JobStateRunning)) + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(rivertype.JobStateRunning)) require.NoError(t, err) require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) @@ -1583,11 +1600,70 @@ func Test_Client_JobList(t *testing.T) { res, err := client.JobList(ctx, nil) require.NoError(t, err) - // sort order is switched by ScheduledAt values: - require.Equal(t, []int64{job2.ID, job3.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + // sort order defaults to ID + require.Equal(t, []int64{job1.ID, job2.ID, job3.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + }) + + t.Run("PaginatesWithAfter_JobListOrderByID", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + job3 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{}) + + res, err := client.JobList(ctx, NewJobListParams().After(JobListCursorFromJob(job1))) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job3.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByID, res.LastCursor.sortField) + require.Equal(t, job3.ID, res.LastCursor.id) + + // No more results + res, err = client.JobList(ctx, NewJobListParams().After(JobListCursorFromJob(job3))) + require.NoError(t, err) + require.Equal(t, []int64{}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Nil(t, res.LastCursor) + + // Descending + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByID, SortOrderDesc).After(JobListCursorFromJob(job3))) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByID, res.LastCursor.sortField) + require.Equal(t, job1.ID, res.LastCursor.id) }) - t.Run("PaginatesWithAfter", func(t *testing.T) { + t.Run("PaginatesWithAfter_JobListOrderByScheduledAt", func(t *testing.T) { + t.Parallel() + + client, bundle := setup(t) + + now := time.Now().UTC() + job1 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{ScheduledAt: &now}) + job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{ScheduledAt: ptrutil.Ptr(now.Add(1 * time.Second))}) + job3 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{ScheduledAt: ptrutil.Ptr(now.Add(2 * time.Second))}) + + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderAsc).After(JobListCursorFromJob(job1))) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job3.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByScheduledAt, res.LastCursor.sortField) + require.Equal(t, job3.ID, res.LastCursor.id) + + // No more results + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderAsc).After(JobListCursorFromJob(job3))) + require.NoError(t, err) + require.Equal(t, []int64{}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Nil(t, res.LastCursor) + + // Descending + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderDesc).After(JobListCursorFromJob(job3))) + require.NoError(t, err) + require.Equal(t, []int64{job2.ID, job1.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByScheduledAt, res.LastCursor.sortField) + require.Equal(t, job1.ID, res.LastCursor.id) + }) + + t.Run("PaginatesWithAfter_JobListOrderByTime", func(t *testing.T) { t.Parallel() client, bundle := setup(t) @@ -1600,26 +1676,23 @@ func Test_Client_JobList(t *testing.T) { job5 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCompleted), ScheduledAt: ptrutil.Ptr(now.Add(-7 * time.Second)), FinalizedAt: ptrutil.Ptr(now.Add(-5 * time.Second))}) job6 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCompleted), ScheduledAt: ptrutil.Ptr(now.Add(-7 * time.Second)), FinalizedAt: &now}) - res, err := client.JobList(ctx, NewJobListParams().States(rivertype.JobStateAvailable).After(JobListCursorFromJob(job1, JobListOrderByTime))) + res, err := client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(rivertype.JobStateAvailable).After(JobListCursorFromJob(job1))) require.NoError(t, err) require.Equal(t, []int64{job2.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByTime, res.LastCursor.sortField) require.Equal(t, job2.ID, res.LastCursor.id) - res, err = client.JobList(ctx, NewJobListParams().States(rivertype.JobStateRunning).After(JobListCursorFromJob(job3, JobListOrderByTime))) + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(rivertype.JobStateRunning).After(JobListCursorFromJob(job3))) require.NoError(t, err) require.Equal(t, []int64{job4.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByTime, res.LastCursor.sortField) require.Equal(t, job4.ID, res.LastCursor.id) - res, err = client.JobList(ctx, NewJobListParams().States(rivertype.JobStateCompleted).After(JobListCursorFromJob(job5, JobListOrderByTime))) + res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc).States(rivertype.JobStateCompleted).After(JobListCursorFromJob(job5))) require.NoError(t, err) require.Equal(t, []int64{job6.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) + require.Equal(t, JobListOrderByTime, res.LastCursor.sortField) require.Equal(t, job6.ID, res.LastCursor.id) - - res, err = client.JobList(ctx, NewJobListParams().OrderBy(JobListOrderByScheduledAt, SortOrderAsc).After(JobListCursorFromJob(job4, JobListOrderByScheduledAt))) - require.NoError(t, err) - require.Equal(t, []int64{job1.ID, job3.ID, job2.ID}, sliceutil.Map(res.Jobs, func(job *rivertype.JobRow) int64 { return job.ID })) - require.Equal(t, JobListOrderByScheduledAt, res.LastCursor.sortField) - require.Equal(t, job2.ID, res.LastCursor.id) }) t.Run("MetadataOnly", func(t *testing.T) { diff --git a/job_list_params.go b/job_list_params.go index 77b08c0a..f8637f44 100644 --- a/job_list_params.go +++ b/job_list_params.go @@ -9,6 +9,7 @@ import ( "time" "github.com/riverqueue/river/internal/dblist" + "github.com/riverqueue/river/internal/util/ptrutil" "github.com/riverqueue/river/rivertype" ) @@ -16,33 +17,48 @@ import ( // job list query. type JobListCursor struct { id int64 + job *rivertype.JobRow // used for JobListCursorFromJob path; not serialized kind string queue string sortField JobListOrderByField - time time.Time + time time.Time // may be empty } // JobListCursorFromJob creates a JobListCursor from a JobRow. -func JobListCursorFromJob(job *rivertype.JobRow, sortField JobListOrderByField) *JobListCursor { - time := job.CreatedAt - switch sortField { +func JobListCursorFromJob(job *rivertype.JobRow) *JobListCursor { + // Other fields are initialized when the cursor is used in After below. + return &JobListCursor{job: job} +} + +func jobListCursorFromJobAndParams(job *rivertype.JobRow, listParams *JobListParams) *JobListCursor { + // A pointer so that we can detect a condition where we accidentally left + // this value unset. + var cursorTime *time.Time + + // Don't include a `default` so `exhaustive` lint can detect omissions. + switch listParams.sortField { + case JobListOrderByID: + cursorTime = ptrutil.Ptr(time.Time{}) case JobListOrderByTime: - time = jobListTimeValue(job) + cursorTime = ptrutil.Ptr(jobListTimeValue(job)) case JobListOrderByFinalizedAt: if job.FinalizedAt != nil { - time = *job.FinalizedAt + cursorTime = job.FinalizedAt } case JobListOrderByScheduledAt: - time = job.ScheduledAt - default: + cursorTime = &job.ScheduledAt + } + + if cursorTime == nil { panic("invalid sort field") } + return &JobListCursor{ id: job.ID, kind: job.Kind, queue: job.Queue, - sortField: sortField, - time: time, + sortField: listParams.sortField, + time: *cursorTime, } } @@ -73,6 +89,10 @@ func (c *JobListCursor) UnmarshalText(text []byte) error { // MarshalText implements encoding.TextMarshaler to encode the cursor as an // opaque string. func (c JobListCursor) MarshalText() ([]byte, error) { + if c.job != nil { + return nil, errors.New("cursor initialized with only a job can't be marshaled; try a cursor from JobListResult instead") + } + wrapperValue := jobListPaginationCursorJSON{ ID: c.id, Kind: c.kind, @@ -103,6 +123,7 @@ type SortOrder int const ( // SortOrderAsc specifies that the sort should in ascending order. SortOrderAsc SortOrder = iota + // SortOrderDesc specifies that the sort should in descending order. SortOrderDesc ) @@ -111,16 +132,30 @@ const ( type JobListOrderByField string const ( + // JobListOrderByID specifies that the sort should be by job ID. + JobListOrderByID JobListOrderByField = "id" + // JobListOrderByFinalizedAt specifies that the sort should be by - // finalized_at. + // `finalized_at`. // // This option must be used in conjunction with filtering by only finalized // job states. JobListOrderByFinalizedAt JobListOrderByField = "finalized_at" - // JobListOrderByScheduledAt specifies that the sort should be by scheduled_at. + + // JobListOrderByScheduledAt specifies that the sort should be by + // `scheduled_at`. JobListOrderByScheduledAt JobListOrderByField = "scheduled_at" - // JobListOrderByTime specifies that the sort should be by time. The specific - // time field used will vary by the first specified job state. + + // JobListOrderByTime specifies that the sort should be by the "best fit" + // time field based on listed state. The best fit is determined by looking + // at the first value given to JobListParams.States. If multiple states are + // specified, the ones after the first will be ignored. + // + // The specific time field used for sorting depends on requested state: + // + // * States `available`, `retryable`, or `scheduled` use `scheduled_at`. + // * State `running` uses `attempted_at`. + // * States `cancelled`, `completed`, or `discarded` use `finalized_at`. JobListOrderByTime JobListOrderByField = "time" ) @@ -146,7 +181,7 @@ type JobListParams struct { func NewJobListParams() *JobListParams { return &JobListParams{ paginationCount: 100, - sortField: JobListOrderByTime, + sortField: JobListOrderByID, sortOrder: SortOrderAsc, states: []rivertype.JobState{ rivertype.JobStateAvailable, @@ -178,7 +213,7 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) { conditionsBuilder := &strings.Builder{} conditions := make([]string, 0, 10) namedArgs := make(map[string]any) - orderBy := []dblist.JobListOrderBy{} + orderBy := make([]dblist.JobListOrderBy, 0, 2) var sortOrder dblist.SortOrder switch p.sortOrder { @@ -208,16 +243,20 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) { } var timeField string - if len(p.states) > 0 && p.sortField == JobListOrderByTime { + switch { + case p.sortField == JobListOrderByID: + // no time field + + case len(p.states) > 0 && p.sortField == JobListOrderByTime: timeField = jobListTimeFieldForState(p.states[0]) - } else { + orderBy = append(orderBy, dblist.JobListOrderBy{Expr: timeField, Order: sortOrder}) + + default: timeField = string(p.sortField) + orderBy = append(orderBy, dblist.JobListOrderBy{Expr: timeField, Order: sortOrder}) } - orderBy = append(orderBy, []dblist.JobListOrderBy{ - {Expr: timeField, Order: sortOrder}, - {Expr: "id", Order: sortOrder}, - }...) + orderBy = append(orderBy, dblist.JobListOrderBy{Expr: "id", Order: sortOrder}) if p.metadataFragment != "" { conditions = append(conditions, `metadata @> @metadata_fragment::jsonb`) @@ -225,12 +264,20 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) { } if p.after != nil { - if sortOrder == dblist.SortOrderAsc { - conditions = append(conditions, fmt.Sprintf(`("%s" > @cursor_time OR ("%s" = @cursor_time AND "id" > @after_id))`, timeField, timeField)) + if p.after.time.IsZero() { // order by ID only + if sortOrder == dblist.SortOrderAsc { + conditions = append(conditions, "(id > @after_id)") + } else { + conditions = append(conditions, "(id < @after_id)") + } } else { - conditions = append(conditions, fmt.Sprintf(`("%s" < @cursor_time OR ("%s" = @cursor_time AND "id" < @after_id))`, timeField, timeField)) + if sortOrder == dblist.SortOrderAsc { + conditions = append(conditions, fmt.Sprintf(`("%s" > @cursor_time OR ("%s" = @cursor_time AND "id" > @after_id))`, timeField, timeField)) + } else { + conditions = append(conditions, fmt.Sprintf(`("%s" < @cursor_time OR ("%s" = @cursor_time AND "id" < @after_id))`, timeField, timeField)) + } + namedArgs["cursor_time"] = p.after.time } - namedArgs["cursor_time"] = p.after.time namedArgs["after_id"] = p.after.id } @@ -259,7 +306,12 @@ func (p *JobListParams) toDBParams() (*dblist.JobListParams, error) { // after the given cursor. func (p *JobListParams) After(cursor *JobListCursor) *JobListParams { result := p.copy() - result.after = cursor + + if cursor.job == nil { + result.after = cursor + } else { + result.after = jobListCursorFromJobAndParams(cursor.job, result) + } return result } @@ -311,7 +363,7 @@ func (p *JobListParams) Queues(queues ...string) *JobListParams { func (p *JobListParams) OrderBy(field JobListOrderByField, direction SortOrder) *JobListParams { result := p.copy() switch field { - case JobListOrderByTime, JobListOrderByScheduledAt: + case JobListOrderByID, JobListOrderByTime, JobListOrderByScheduledAt: result.sortField = field case JobListOrderByFinalizedAt: result.sortField = field @@ -341,6 +393,7 @@ func (p *JobListParams) States(states ...rivertype.JobState) *JobListParams { } func jobListTimeFieldForState(state rivertype.JobState) string { + // Don't include a `default` so `exhaustive` lint can detect omissions. switch state { case rivertype.JobStateAvailable, rivertype.JobStateRetryable, rivertype.JobStateScheduled: return "scheduled_at" @@ -348,28 +401,31 @@ func jobListTimeFieldForState(state rivertype.JobState) string { return "attempted_at" case rivertype.JobStateCancelled, rivertype.JobStateCompleted, rivertype.JobStateDiscarded: return "finalized_at" - default: - return "created_at" // should never happen } + + return "created_at" // should never happen } func jobListTimeValue(job *rivertype.JobRow) time.Time { + // Don't include a `default` so `exhaustive` lint can detect omissions. switch job.State { case rivertype.JobStateAvailable, rivertype.JobStateRetryable, rivertype.JobStateScheduled: return job.ScheduledAt + case rivertype.JobStateRunning: if job.AttemptedAt == nil { // This should never happen unless a job has been manually manipulated. return job.CreatedAt } return *job.AttemptedAt + case rivertype.JobStateCancelled, rivertype.JobStateCompleted, rivertype.JobStateDiscarded: if job.FinalizedAt == nil { // This should never happen unless a job has been manually manipulated. return job.CreatedAt } return *job.FinalizedAt - default: - return job.CreatedAt // should never happen } + + return job.CreatedAt // should never happen } diff --git a/job_list_params_test.go b/job_list_params_test.go index f8533167..b3866fa1 100644 --- a/job_list_params_test.go +++ b/job_list_params_test.go @@ -15,6 +15,42 @@ import ( func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { t.Parallel() + jobRow := &rivertype.JobRow{ + ID: 4, + Kind: "test", + Queue: "test", + State: rivertype.JobStateRunning, + } + + cursor := JobListCursorFromJob(jobRow) + require.Zero(t, cursor.id) + require.Equal(t, jobRow, cursor.job) + require.Zero(t, cursor.kind) + require.Zero(t, cursor.queue) + require.Zero(t, cursor.sortField) + require.Zero(t, cursor.time) +} + +func Test_JobListCursor_jobListCursorFromJobAndParams(t *testing.T) { + t.Parallel() + + t.Run("OrderByID", func(t *testing.T) { + t.Parallel() + + jobRow := &rivertype.JobRow{ + ID: 4, + Kind: "test", + Queue: "test", + State: rivertype.JobStateRunning, + } + + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().After(JobListCursorFromJob(jobRow))) + require.Equal(t, jobRow.ID, cursor.id) + require.Equal(t, jobRow.Kind, cursor.kind) + require.Equal(t, jobRow.Queue, cursor.queue) + require.Zero(t, cursor.time) + }) + for i, state := range []rivertype.JobState{ rivertype.JobStateAvailable, rivertype.JobStateRetryable, @@ -22,7 +58,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { } { i, state := i, state - t.Run(fmt.Sprintf("ScheduledAtUsedFor%sJob", state), func(t *testing.T) { + t.Run(fmt.Sprintf("OrderByTimeScheduledAtUsedFor%sJob", state), func(t *testing.T) { t.Parallel() now := time.Now().UTC() @@ -35,7 +71,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { ScheduledAt: now.Add(-10 * time.Second), } - cursor := JobListCursorFromJob(jobRow, JobListOrderByTime) + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) require.Equal(t, jobRow.ID, cursor.id) require.Equal(t, jobRow.Kind, cursor.kind) require.Equal(t, jobRow.Queue, cursor.queue) @@ -50,7 +86,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { } { i, state := i, state - t.Run(fmt.Sprintf("FinalizedAtUsedFor%sJob", state), func(t *testing.T) { + t.Run(fmt.Sprintf("OrderByTimeFinalizedAtUsedFor%sJob", state), func(t *testing.T) { t.Parallel() now := time.Now().UTC() @@ -65,7 +101,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { ScheduledAt: now.Add(-10 * time.Second), } - cursor := JobListCursorFromJob(jobRow, JobListOrderByTime) + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) require.Equal(t, jobRow.ID, cursor.id) require.Equal(t, jobRow.Kind, cursor.kind) require.Equal(t, jobRow.Queue, cursor.queue) @@ -73,7 +109,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { }) } - t.Run("RunningJobUsesAttemptedAt", func(t *testing.T) { + t.Run("OrderByTimeRunningJobUsesAttemptedAt", func(t *testing.T) { t.Parallel() now := time.Now().UTC() @@ -87,14 +123,14 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { ScheduledAt: now.Add(-10 * time.Second), } - cursor := JobListCursorFromJob(jobRow, JobListOrderByTime) + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) require.Equal(t, jobRow.ID, cursor.id) require.Equal(t, jobRow.Kind, cursor.kind) require.Equal(t, jobRow.Queue, cursor.queue) require.Equal(t, *jobRow.AttemptedAt, cursor.time) }) - t.Run("UnknownJobStateUsesCreatedAt", func(t *testing.T) { + t.Run("OrderByTimeUnknownJobStateUsesCreatedAt", func(t *testing.T) { t.Parallel() now := time.Now().UTC() @@ -107,7 +143,7 @@ func Test_JobListCursor_JobListCursorFromJob(t *testing.T) { ScheduledAt: now.Add(-10 * time.Second), } - cursor := JobListCursorFromJob(jobRow, JobListOrderByTime) + cursor := jobListCursorFromJobAndParams(jobRow, NewJobListParams().OrderBy(JobListOrderByTime, SortOrderAsc)) require.Equal(t, jobRow.ID, cursor.id) require.Equal(t, jobRow.Kind, cursor.kind) require.Equal(t, jobRow.Queue, cursor.queue) @@ -122,20 +158,36 @@ func Test_JobListCursor_MarshalJSON(t *testing.T) { t.Parallel() now := time.Now().UTC() - params := &JobListCursor{ + cursor := &JobListCursor{ id: 4, kind: "test_kind", queue: "test_queue", time: now, } - text, err := json.Marshal(params) + text, err := json.Marshal(cursor) require.NoError(t, err) require.NotEqual(t, "", text) unmarshaledParams := &JobListCursor{} require.NoError(t, json.Unmarshal(text, unmarshaledParams)) - require.Equal(t, params, unmarshaledParams) + require.Equal(t, cursor, unmarshaledParams) + }) + + t.Run("ErrorsOnJobOnlyCursor", func(t *testing.T) { + t.Parallel() + + jobRow := &rivertype.JobRow{ + ID: 4, + Kind: "test", + Queue: "test", + State: rivertype.JobStateRunning, + } + + cursor := JobListCursorFromJob(jobRow) + + _, err := json.Marshal(cursor) + require.EqualError(t, err, "json: error calling MarshalText for type *river.JobListCursor: cursor initialized with only a job can't be marshaled; try a cursor from JobListResult instead") }) }