Use configured time for job snoozes and retries

[pablobaldez]

Config.Test.Time is used when River evaluates whether scheduled jobs
are eligible to run, but JobSnooze and the default retry policy derived
their next scheduled_at values from the process wall clock. When a
configured time generator differed from wall time, this mixed two
timelines in the same scheduling flow and could make snoozed or retried
jobs run too early or remain ineligible for much longer than intended.

Here, derive snooze timestamps from the executor's configured time and
initialize River's default retry policy from the same base service clock.
This also initializes the default fallback policy in producer and
rivertest.Worker execution paths. Custom retry policies are unchanged,
and zero-value use continues to fall back to wall time.

Regression coverage exercises both job snoozing and retrying through the
public client boundary with a configured time generator. Focused tests
and make lint pass. The full local make test run encounters existing
TestProducer_PollOnly timeouts that reproduce on the unchanged base
commit.

Fixes #1322.

[brandur]

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.

[brandur]

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?