Skip to content

Use configured time for job snoozes and retries #1332

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

Open
pablobaldez wants to merge 1 commit into
base: master
Choose a base branch
from
+83 −3
Open

Use configured time for job snoozes and retries #1332

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
67 changes: 67 additions & 0 deletions client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -994,6 +994,40 @@ func Test_Client_Common(t *testing.T) {
require.WithinDuration(t, time.Now(), *updatedJob.FinalizedAt, 2*time.Second)
})

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

config, bundle := setupConfig(t)
configuredNow := time.Now().UTC().Add(-10 * time.Minute).Truncate(time.Microsecond)
timeStub := &riversharedtest.TimeStub{}
timeStub.StubNow(configuredNow)
config.Test.Time = timeStub
client := newTestClient(t, bundle.dbPool, config)

type JobArgs struct {
testutil.JobArgsReflectKind[JobArgs]
}

AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
return errors.New("retry using configured time")
}))

subscribeChan := subscribe(t, client)
startClient(ctx, t, client)

insertRes, err := client.Insert(ctx, &JobArgs{}, nil)
require.NoError(t, err)

event := riversharedtest.WaitOrTimeout(t, subscribeChan)
require.Equal(t, EventKindJobFailed, event.Kind)
require.Equal(t, rivertype.JobStateRetryable, event.Job.State)
require.WithinDuration(t, configuredNow.Add(time.Second), event.Job.ScheduledAt, 150*time.Millisecond)

updatedJob, err := client.JobGet(ctx, insertRes.Job.ID)
require.NoError(t, err)
require.WithinDuration(t, configuredNow.Add(time.Second), updatedJob.ScheduledAt, 150*time.Millisecond)
})

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

Expand Down Expand Up @@ -1025,6 +1059,39 @@ func Test_Client_Common(t *testing.T) {
require.WithinDuration(t, time.Now().Add(15*time.Minute), updatedJob.ScheduledAt, 2*time.Second)
})

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

config, bundle := setupConfig(t)
configuredNow := time.Now().UTC().Add(-10 * time.Minute).Truncate(time.Microsecond)
timeStub := &riversharedtest.TimeStub{}
timeStub.StubNow(configuredNow)
config.Test.Time = timeStub
client := newTestClient(t, bundle.dbPool, config)

type JobArgs struct {
testutil.JobArgsReflectKind[JobArgs]
}

AddWorker(client.config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
return JobSnooze(15 * time.Minute)
}))

subscribeChan := subscribe(t, client)
startClient(ctx, t, client)

insertRes, err := client.Insert(ctx, &JobArgs{}, nil)
require.NoError(t, err)

event := riversharedtest.WaitOrTimeout(t, subscribeChan)
require.Equal(t, EventKindJobSnoozed, event.Kind)
require.Equal(t, configuredNow.Add(15*time.Minute), event.Job.ScheduledAt)

updatedJob, err := client.JobGet(ctx, insertRes.Job.ID)
require.NoError(t, err)
require.Equal(t, configuredNow.Add(15*time.Minute), updatedJob.ScheduledAt)
})

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

Expand Down
2 changes: 1 addition & 1 deletion internal/jobexecutor/job_executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -383,7 +383,7 @@ func (e *JobExecutor) reportResult(ctx context.Context, jobRow *rivertype.JobRow
slog.String("job_kind", jobRow.Kind),
slog.Duration("duration", snoozeErr.Duration),
)
nextAttemptScheduledAt := time.Now().Add(snoozeErr.Duration)
nextAttemptScheduledAt := e.Time.Now().Add(snoozeErr.Duration)

snoozesValue := gjson.GetBytes(jobRow.Metadata, "snoozes").Int()
if res.MetadataUpdates == nil {
Expand Down
4 changes: 3 additions & 1 deletion producer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -907,6 +907,8 @@ func (p *producer) heartbeatLogLoop(ctx context.Context, wg *sync.WaitGroup) {
}

func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.JobRow) {
defaultClientRetryPolicy := baseservice.Init(&p.Archetype, &DefaultClientRetryPolicy{})

for _, job := range jobs {
workInfo, ok := p.workers.workersMap[job.Kind]

Expand All @@ -924,7 +926,7 @@ func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.
ClientJobTimeout: p.jobTimeout,
ClientRetryPolicy: p.retryPolicy,
Completer: p.completer,
DefaultClientRetryPolicy: &DefaultClientRetryPolicy{},
DefaultClientRetryPolicy: defaultClientRetryPolicy,
ErrorHandler: p.errorHandler,
PluginLookupByJob: p.config.PluginLookupByJob,
PluginLookupGlobal: p.config.PluginLookupGlobal,
Expand Down
10 changes: 10 additions & 0 deletions retry_policy.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"math/rand/v2"
"time"

"github.com/riverqueue/river/rivershared/baseservice"
"github.com/riverqueue/river/rivershared/util/timeutil"
"github.com/riverqueue/river/rivertype"
)
Expand All @@ -27,9 +28,15 @@ type ClientRetryPolicy interface {

// DefaultClientRetryPolicy is River's default retry policy.
type DefaultClientRetryPolicy struct {
baseService baseservice.BaseService
timeNowFunc func() time.Time
}

// GetBaseService returns the retry policy's base service.
func (p *DefaultClientRetryPolicy) GetBaseService() *baseservice.BaseService {

@brandur brandur Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hey @pablobaldez, ideally we don't expose this internal stuff in River's public API.

I asked Codex on how it might do this differently without exposing extra API and it suggested this:

I’d extract the clock-aware implementation into an internal package:

// internal/retrypolicy/default.go
type Default struct {
    timeGenerator rivertype.TimeGenerator
}

func NewDefault(timeGenerator rivertype.TimeGenerator) *Default {
    return &Default{timeGenerator: timeGenerator}
}

func (p *Default) NextRetry(job *rivertype.JobRow) time.Time {
    return NextRetryAt(p.timeGenerator.Now().UTC(), job)
}

func NextRetryAt(now time.Time, job *rivertype.JobRow) time.Time {
    // Existing retry calculation and jitter.
}

The public zero-value policy remains unchanged and delegates to the shared calculation:

type DefaultClientRetryPolicy struct {
    timeNowFunc func() time.Time
}

func (p *DefaultClientRetryPolicy) NextRetry(job *rivertype.JobRow) time.Time {
    return retrypolicy.NextRetryAt(p.timeNowUTC(), job)
}

Internally, River substitutes the clock-aware implementation when the default policy is selected:

if _, ok := config.RetryPolicy.(*DefaultClientRetryPolicy); ok {
    config.RetryPolicy = retrypolicy.NewDefault(archetype.Time)
}

Producer and rivertest.Worker fallback policies can use the internal constructor directly:

defaultClientRetryPolicy := retrypolicy.NewDefault(archetype.Time)

This would:

  • Remove GetBaseService and the baseservice dependency from the public policy.
  • Preserve zero-value wall-clock behavior for direct external use.
  • Keep Config.Test.Time consistent in Client, producer fallback, and rivertest.
  • Avoid mutating a public policy instance if it’s shared between clients.
  • Add no externally accessible API because internal/retrypolicy cannot be imported outside River.

That seems like a reasonable thing to try. Thoughts?

return &p.baseService
}

// NextRetry gets the next retry given for the given job, accounting for when it
// was last attempted and what attempt number that was. Reschedules using a
// basic exponential backoff of `ATTEMPT^4`, so after the first failure a new
Expand Down Expand Up @@ -61,6 +68,9 @@ func (p *DefaultClientRetryPolicy) timeNowUTC() time.Time {
if p.timeNowFunc != nil {
return p.timeNowFunc()
}
if p.baseService.Time != nil {
return p.baseService.Time.Now().UTC()
}

return time.Now().UTC()
}
Expand Down
3 changes: 2 additions & 1 deletion rivertest/worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,7 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job
pluginlookup.InitBaseServices(archetype, hooks)
pluginlookup.InitBaseServices(archetype, middleware)
pluginlookup.InitBaseServices(archetype, plugins)
defaultClientRetryPolicy := baseservice.Init(archetype, &river.DefaultClientRetryPolicy{})

updatedJobRow, err := exec.JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{
ID: job.ID,
Expand Down Expand Up @@ -192,7 +193,7 @@ func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job
ClientJobTimeout: w.config.JobTimeout,
ClientRetryPolicy: w.config.RetryPolicy,
Completer: completer,
DefaultClientRetryPolicy: &river.DefaultClientRetryPolicy{},
DefaultClientRetryPolicy: defaultClientRetryPolicy,

@brandur brandur Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you move the baseservice.Init invocation inline here? No point in making the code more indirect by defining defaultClientRetryPolicy all the way up there.

ErrorHandler: &errorHandlerWrapper{
HandleErrorFunc: func(ctx context.Context, job *rivertype.JobRow, err error) *jobexecutor.ErrorHandlerResult {
resultErr = err
Expand Down
Loading