Skip to content

Make sure to use timeString/timeStringNullable everywhere for SQLite #1353

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
brandur merged 1 commit into from
Aug 21, 2026
Merged

Make sure to use timeString/timeStringNullable everywhere for SQLite #1353

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Fixed

- SQLite job completion, rescue, and full-update paths now serialize timestamps using River's standard millisecond format instead of relying on database driver serialization, preventing inconsistent timestamp representations from being persisted. [PR #1353](https://github.com/riverqueue/river/pull/1353)

## [0.44.0] - 2026-08-18

### Added
Expand Down
58 changes: 31 additions & 27 deletions riverdriver/riverdrivertest/job_update.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,10 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
}
}

// Deliberately includes sub-millisecond precision so tests can verify that
// drivers normalize timestamps to their declared precision.
precisionTestTime := time.Date(2025, 4, 30, 13, 26, 39, 123400000, time.UTC)

t.Run("JobCancel", func(t *testing.T) {
t.Parallel()

Expand Down Expand Up @@ -147,7 +151,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT

exec, bundle := setup(ctx, t)

now := time.Now().UTC()
now := precisionTestTime

job1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
Metadata: []byte(`{"river:rescue_count": 5, "something": "else"}`),
Expand Down Expand Up @@ -187,15 +191,15 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.NoError(t, err)
require.Equal(t, "message1", updatedJob1.Errors[0].Error)
require.Nil(t, updatedJob1.FinalizedAt)
require.WithinDuration(t, now, updatedJob1.ScheduledAt, bundle.driver.TimePrecision())
require.Equal(t, now.Truncate(bundle.driver.TimePrecision()), updatedJob1.ScheduledAt)
require.Equal(t, rivertype.JobStateAvailable, updatedJob1.State)
require.JSONEq(t, `{"river:rescue_count": 6, "something": "else"}`, string(updatedJob1.Metadata))

updatedJob2, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job2.ID})
require.NoError(t, err)
require.Equal(t, "message2", updatedJob2.Errors[0].Error)
require.WithinDuration(t, now, *updatedJob2.FinalizedAt, bundle.driver.TimePrecision())
require.WithinDuration(t, now, updatedJob2.ScheduledAt, bundle.driver.TimePrecision())
require.Equal(t, now.Truncate(bundle.driver.TimePrecision()), *updatedJob2.FinalizedAt)
require.Equal(t, now.Truncate(bundle.driver.TimePrecision()), updatedJob2.ScheduledAt)
require.Equal(t, rivertype.JobStateDiscarded, updatedJob2.State)
require.JSONEq(t, `{"river:rescue_count": 1}`, string(updatedJob2.Metadata))
})
Expand Down Expand Up @@ -585,9 +589,9 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
t.Run("CompletesARunningJob", func(t *testing.T) {
t.Parallel()

exec, _ := setup(ctx, t)
exec, bundle := setup(ctx, t)

now := time.Now().UTC()
now := precisionTestTime

job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
State: ptrutil.Ptr(rivertype.JobStateRunning),
Expand All @@ -598,7 +602,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.NoError(t, err)
jobAfter := jobsAfter[0]
require.Equal(t, rivertype.JobStateCompleted, jobAfter.State)
require.WithinDuration(t, now, *jobAfter.FinalizedAt, time.Microsecond)
require.Equal(t, now.Truncate(bundle.driver.TimePrecision()), *jobAfter.FinalizedAt)

jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
require.NoError(t, err)
Expand Down Expand Up @@ -670,9 +674,9 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
t.Run("SetsARunningJobToRetryable", func(t *testing.T) {
t.Parallel()

exec, _ := setup(ctx, t)
exec, bundle := setup(ctx, t)

now := time.Now().UTC()
now := precisionTestTime

job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
State: ptrutil.Ptr(rivertype.JobStateRunning),
Expand All @@ -683,7 +687,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.NoError(t, err)
jobAfter := jobsAfter[0]
require.Equal(t, rivertype.JobStateRetryable, jobAfter.State)
require.WithinDuration(t, now, jobAfter.ScheduledAt, time.Microsecond)
require.Equal(t, now.Truncate(bundle.driver.TimePrecision()), jobAfter.ScheduledAt)

jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
require.NoError(t, err)
Expand All @@ -701,7 +705,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
t.Run("SetsAnInterruptedRunningJobToAvailableWithUpdatedAttempt", func(t *testing.T) {
t.Parallel()

exec, _ := setup(ctx, t)
exec, bundle := setup(ctx, t)

now := time.Now().UTC()
attempt := 2
Expand All @@ -720,7 +724,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.Equal(t, attempt, jobAfter.Attempt)
require.Equal(t, rivertype.JobStateAvailable, jobAfter.State)
require.Equal(t, 3, jobAfter.MaxAttempts)
require.WithinDuration(t, now, jobAfter.ScheduledAt, time.Microsecond)
require.WithinDuration(t, now, jobAfter.ScheduledAt, bundle.driver.TimePrecision())

jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
require.NoError(t, err)
Expand Down Expand Up @@ -841,7 +845,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
t.Run("CancelsARunningJob", func(t *testing.T) {
t.Parallel()

exec, _ := setup(ctx, t)
exec, bundle := setup(ctx, t)

now := time.Now().UTC()

Expand All @@ -855,7 +859,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.NoError(t, err)
jobAfter := jobsAfter[0]
require.Equal(t, rivertype.JobStateCancelled, jobAfter.State)
require.WithinDuration(t, now, *jobAfter.FinalizedAt, time.Microsecond)
require.WithinDuration(t, now, *jobAfter.FinalizedAt, bundle.driver.TimePrecision())

jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
require.NoError(t, err)
Expand All @@ -870,7 +874,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
t.Run("DiscardsARunningJob", func(t *testing.T) {
t.Parallel()

exec, _ := setup(ctx, t)
exec, bundle := setup(ctx, t)

now := time.Now().UTC()

Expand All @@ -884,7 +888,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.NoError(t, err)
jobAfter := jobsAfter[0]
require.Equal(t, rivertype.JobStateDiscarded, jobAfter.State)
require.WithinDuration(t, now, *jobAfter.FinalizedAt, time.Microsecond)
require.WithinDuration(t, now, *jobAfter.FinalizedAt, bundle.driver.TimePrecision())
require.Equal(t, "unique-key", string(jobAfter.UniqueKey))
require.Equal(t, rivertype.JobStates(), jobAfter.UniqueStates)

Expand All @@ -900,7 +904,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
t.Run("SnoozesARunningJob_WithNoPreexistingMetadata", func(t *testing.T) {
t.Parallel()

exec, _ := setup(ctx, t)
exec, bundle := setup(ctx, t)

now := time.Now().UTC()
snoozeUntil := now.Add(1 * time.Minute)
Expand All @@ -918,7 +922,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.Equal(t, job.MaxAttempts, jobAfter.MaxAttempts)
require.JSONEq(t, `{"snoozes": 1}`, string(jobAfter.Metadata))
require.Equal(t, rivertype.JobStateScheduled, jobAfter.State)
require.WithinDuration(t, snoozeUntil, jobAfter.ScheduledAt, time.Microsecond)
require.WithinDuration(t, snoozeUntil, jobAfter.ScheduledAt, bundle.driver.TimePrecision())

jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
require.NoError(t, err)
Expand All @@ -932,7 +936,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
t.Run("SnoozesARunningJob_WithPreexistingMetadata", func(t *testing.T) {
t.Parallel()

exec, _ := setup(ctx, t)
exec, bundle := setup(ctx, t)

now := time.Now().UTC()
snoozeUntil := now.Add(1 * time.Minute)
Expand All @@ -951,7 +955,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.Equal(t, job.MaxAttempts, jobAfter.MaxAttempts)
require.JSONEq(t, `{"foo": "bar", "snoozes": 6}`, string(jobAfter.Metadata))
require.Equal(t, rivertype.JobStateScheduled, jobAfter.State)
require.WithinDuration(t, snoozeUntil, jobAfter.ScheduledAt, time.Microsecond)
require.WithinDuration(t, snoozeUntil, jobAfter.ScheduledAt, bundle.driver.TimePrecision())

jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
require.NoError(t, err)
Expand All @@ -966,7 +970,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
t.Run("JobSetStateIfRunningMany_MultipleJobsAtOnce", func(t *testing.T) {
t.Parallel()

exec, _ := setup(ctx, t)
exec, bundle := setup(ctx, t)

now := time.Now().UTC()
future := now.Add(10 * time.Second)
Expand All @@ -985,12 +989,12 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
require.Len(t, jobsAfter, 3)
completedJob := jobsAfter[0]
require.Equal(t, rivertype.JobStateCompleted, completedJob.State)
require.WithinDuration(t, now, *completedJob.FinalizedAt, time.Microsecond)
require.WithinDuration(t, now, *completedJob.FinalizedAt, bundle.driver.TimePrecision())
require.JSONEq(t, `{"a":"b"}`, string(completedJob.Metadata))

retryableJob := jobsAfter[1]
require.Equal(t, rivertype.JobStateRetryable, retryableJob.State)
require.WithinDuration(t, future, retryableJob.ScheduledAt, time.Microsecond)
require.WithinDuration(t, future, retryableJob.ScheduledAt, bundle.driver.TimePrecision())
// validate error payload:
require.Len(t, retryableJob.Errors, 1)
require.Equal(t, now, retryableJob.Errors[0].At)
Expand All @@ -1000,7 +1004,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT

cancelledJob := jobsAfter[2]
require.Equal(t, rivertype.JobStateCancelled, cancelledJob.State)
require.WithinDuration(t, now, *cancelledJob.FinalizedAt, time.Microsecond)
require.WithinDuration(t, now, *cancelledJob.FinalizedAt, bundle.driver.TimePrecision())
})

t.Run("JobUpdate", func(t *testing.T) {
Expand Down Expand Up @@ -1051,7 +1055,7 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT

job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{})

now := time.Now().UTC()
now := precisionTestTime

updatedJob, err := exec.JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{
ID: job.ID,
Expand All @@ -1074,10 +1078,10 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
})
require.NoError(t, err)
require.Equal(t, 7, updatedJob.Attempt)
require.WithinDuration(t, now, *updatedJob.AttemptedAt, bundle.driver.TimePrecision())
require.Equal(t, now.Truncate(bundle.driver.TimePrecision()), *updatedJob.AttemptedAt)
require.Equal(t, []string{"worker1"}, updatedJob.AttemptedBy)
require.Equal(t, "message", updatedJob.Errors[0].Error)
require.WithinDuration(t, now, *updatedJob.FinalizedAt, bundle.driver.TimePrecision())
require.Equal(t, now.Truncate(bundle.driver.TimePrecision()), *updatedJob.FinalizedAt)
require.Equal(t, 99, updatedJob.MaxAttempts)
require.JSONEq(t, `{"foo":"bar"}`, string(updatedJob.Metadata))
require.Equal(t, rivertype.JobStateDiscarded, updatedJob.State)
Expand Down
10 changes: 5 additions & 5 deletions riverdriver/riversqlite/internal/dbsqlc/river_job.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -489,7 +489,7 @@ UPDATE /* TEMPLATE: schema */river_job
SET
errors = jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(@error))),
finalized_at = cast(sqlc.narg('finalized_at') as text),
scheduled_at = @scheduled_at,
scheduled_at = cast(@scheduled_at AS text),
metadata = jsonb_set(
metadata,
'$."river:rescue_count"',
Expand Down Expand Up @@ -606,13 +606,13 @@ SET
finalized_at = CASE WHEN /* should_cancel */((@state = 'available' OR @state = 'retryable' OR @state = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
THEN coalesce(cast(sqlc.narg('now') AS text), datetime('now', 'subsec'))
WHEN cast(@finalized_at_do_update AS boolean)
THEN @finalized_at
THEN cast(sqlc.narg('finalized_at') AS text)
ELSE finalized_at END,
metadata = CASE WHEN cast(@metadata_do_merge AS boolean)
THEN jsonb_patch(json(metadata), json(@metadata_updates))
ELSE metadata END,
scheduled_at = 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(@scheduled_at_do_update AS boolean)
THEN @scheduled_at
THEN cast(@scheduled_at AS text)
ELSE scheduled_at END,
state = CASE WHEN /* should_cancel */((@state = 'available' OR @state = 'retryable' OR @state = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
THEN 'cancelled'
Expand All @@ -634,10 +634,10 @@ RETURNING *;
UPDATE /* TEMPLATE: schema */river_job
SET
attempt = CASE WHEN cast(@attempt_do_update AS boolean) THEN @attempt ELSE attempt END,
attempted_at = CASE WHEN cast(@attempted_at_do_update AS boolean) THEN @attempted_at ELSE attempted_at END,
attempted_at = CASE WHEN cast(@attempted_at_do_update AS boolean) THEN cast(sqlc.narg('attempted_at') AS text) ELSE attempted_at END,
attempted_by = CASE WHEN cast(@attempted_by_do_update AS boolean) THEN jsonb(@attempted_by) ELSE attempted_by END,
errors = CASE WHEN cast(@errors_do_update AS boolean) THEN jsonb(@errors) ELSE errors END,
finalized_at = CASE WHEN cast(@finalized_at_do_update AS boolean) THEN @finalized_at ELSE finalized_at END,
finalized_at = CASE WHEN cast(@finalized_at_do_update AS boolean) THEN cast(sqlc.narg('finalized_at') AS text) ELSE finalized_at END,
max_attempts = CASE WHEN cast(@max_attempts_do_update AS boolean) THEN @max_attempts ELSE max_attempts END,
metadata = CASE WHEN cast(@metadata_do_update AS boolean) THEN jsonb(@metadata) ELSE metadata END,
state = CASE WHEN cast(@state_do_update AS boolean) THEN @state ELSE state END
Expand Down
21 changes: 10 additions & 11 deletions riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading
Loading