diff --git a/CHANGELOG.md b/CHANGELOG.md index 0755bb469..87c94eaa2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Refresh a durable periodic job's saved row only after its next run time is loaded at startup or committed. When a startup write rolled back, keep-alive could revive an expired row, and the next leader enqueued every run since the row's old next run time. [PR #1540](https://github.com/riverqueue/river/pull/1540). - Fail retries that omit a saved cursor-only step, while preserving their checkpoints for recovery. Jobs with an existing step marker retain their behavior, including legacy stale cursor entries. [PR #1534](https://github.com/riverqueue/river/pull/1534). - Fixed SQLite metadata updates recursively merging nested objects and deleting keys set to JSON null. Updates now replace whole top-level values, matching Postgres and preventing completed resumable cursor steps from running again on later retries. [PR #1533](https://github.com/riverqueue/river/pull/1533). diff --git a/internal/maintenance/periodic_job_enqueuer.go b/internal/maintenance/periodic_job_enqueuer.go index bbb0f808b..aa67a4695 100644 --- a/internal/maintenance/periodic_job_enqueuer.go +++ b/internal/maintenance/periodic_job_enqueuer.go @@ -55,6 +55,7 @@ type PeriodicJob struct { RunOnStart bool ScheduleFunc func(time.Time) time.Time + keepAlive bool // true once nextRunAt is loaded from the pilot or committed in this leadership term nextRunAt time.Time // set on service start } @@ -197,6 +198,7 @@ func (s *PeriodicJobEnqueuer) AddSafely(periodicJob *PeriodicJob) (rivertype.Per return 0, err } + periodicJob.keepAlive = false s.periodicJobs[handle] = periodicJob s.nextHandle++ @@ -228,6 +230,7 @@ func (s *PeriodicJobEnqueuer) AddManySafely(periodicJobs []*PeriodicJob) ([]rive return nil, err } + periodicJob.keepAlive = false s.periodicJobs[handles[i]] = periodicJob s.nextHandle++ } @@ -319,8 +322,9 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { s.StaggerStart(ctx) var ( - initialPeriodicJobs []*riverpilot.PeriodicJob - subServices []startstop.Service + initialPeriodicJobs []*riverpilot.PeriodicJob + initialPeriodicJobsMap map[string]time.Time + subServices []startstop.Service ) if err := func() error { var err error @@ -341,6 +345,14 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { } } + initialPeriodicJobsMap = sliceutil.KeyBy(initialPeriodicJobs, + func(j *riverpilot.PeriodicJob) (string, time.Time) { return j.ID, j.NextRunAt }) + s.mu.Lock() + for _, periodicJob := range s.periodicJobs { + _, periodicJob.keepAlive = initialPeriodicJobsMap[periodicJob.ID] + } + s.mu.Unlock() + subServices = []startstop.Service{ startstop.StartStopFunc(s.periodicJobKeepAliveAndReapPeriodically), } @@ -375,62 +387,61 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { default: } - // Initial set of periodic job IDs mapped to next run at times fetched - // from a configured pilot. Not used in most cases. - initialPeriodicJobsMap := sliceutil.KeyBy(initialPeriodicJobs, - func(j *riverpilot.PeriodicJob) (string, time.Time) { return j.ID, j.NextRunAt }) - var lastHandleSeen rivertype.PeriodicJobHandle = -1 // so handle 0 is considered validateInsertRunOnStartAndScheduleNewlyAdded := func() { - s.mu.RLock() - defer s.mu.RUnlock() - var ( insertParamsMany []*rivertype.JobInsertParams now = s.Time.Now() periodicJobUpsertParams = &riverpilot.PeriodicJobUpsertManyParams{Schema: s.Config.Schema} + periodicJobsToKeepAlive = make(map[rivertype.PeriodicJobHandle]*PeriodicJob) ) - // Handle periodic jobs in sorted order so we can correctly account - // for the most recently added one that we've seen. - sortedPeriodicJobHandles := maputil.Keys(s.periodicJobs) - slices.Sort(sortedPeriodicJobHandles) + func() { + s.mu.RLock() + defer s.mu.RUnlock() - for _, handle := range sortedPeriodicJobHandles { - if handle <= lastHandleSeen { - continue - } + // Handle periodic jobs in sorted order so we can correctly account + // for the most recently added one that we've seen. + sortedPeriodicJobHandles := maputil.Keys(s.periodicJobs) + slices.Sort(sortedPeriodicJobHandles) - lastHandleSeen = handle + for _, handle := range sortedPeriodicJobHandles { + if handle <= lastHandleSeen { + continue + } - periodicJob := s.periodicJobs[handle].mustValidate() + lastHandleSeen = handle - if nextRunAt, ok := initialPeriodicJobsMap[periodicJob.ID]; periodicJob.ID != "" && ok { - periodicJob.nextRunAt = nextRunAt - delete(initialPeriodicJobsMap, periodicJob.ID) - } else { - periodicJob.nextRunAt = periodicJob.ScheduleFunc(now) - } + periodicJob := s.periodicJobs[handle].mustValidate() - if periodicJob.ID != "" { - periodicJobUpsertParams.Jobs = append(periodicJobUpsertParams.Jobs, &riverpilot.PeriodicJobUpsertParams{ - ID: periodicJob.ID, - NextRunAt: periodicJob.nextRunAt, - UpdatedAt: s.Time.Now(), - }) - } + if nextRunAt, ok := initialPeriodicJobsMap[periodicJob.ID]; periodicJob.ID != "" && ok { + periodicJob.nextRunAt = nextRunAt + delete(initialPeriodicJobsMap, periodicJob.ID) + } else { + periodicJob.nextRunAt = periodicJob.ScheduleFunc(now) + } - if !periodicJob.RunOnStart { - continue - } + if periodicJob.ID != "" { + periodicJobsToKeepAlive[handle] = periodicJob + periodicJobUpsertParams.Jobs = append(periodicJobUpsertParams.Jobs, &riverpilot.PeriodicJobUpsertParams{ + ID: periodicJob.ID, + NextRunAt: periodicJob.nextRunAt, + UpdatedAt: s.Time.Now(), + }) + } + + if !periodicJob.RunOnStart { + continue + } - if insertParams, ok := s.insertParamsFromConstructor(ctx, periodicJob.ID, periodicJob.ConstructorFunc, now); ok { - insertParamsMany = append(insertParamsMany, insertParams) + if insertParams, ok := s.insertParamsFromConstructor(ctx, periodicJob.ID, periodicJob.ConstructorFunc, now); ok { + insertParamsMany = append(insertParamsMany, insertParams) + } } - } + }() - s.insertBatch(ctx, insertParamsMany, periodicJobUpsertParams) + s.insertBatch(ctx, insertParamsMany, periodicJobUpsertParams, periodicJobsToKeepAlive) if len(insertParamsMany) > 0 { s.Logger.DebugContext(ctx, s.Name+": Inserted RunOnStart jobs", "num_jobs", len(insertParamsMany)) @@ -450,6 +461,7 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { var ( insertParamsMany []*rivertype.JobInsertParams periodicJobUpsertParams = &riverpilot.PeriodicJobUpsertManyParams{Schema: s.Config.Schema} + periodicJobsToKeepAlive = make(map[rivertype.PeriodicJobHandle]*PeriodicJob) ) now := s.Time.Now() @@ -463,7 +475,7 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { s.mu.RLock() defer s.mu.RUnlock() - for _, periodicJob := range s.periodicJobs { + for handle, periodicJob := range s.periodicJobs { if periodicJob.nextRunAt.IsZero() || !periodicJob.nextRunAt.Before(nowWithMargin) { continue } @@ -479,6 +491,7 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { periodicJob.nextRunAt = periodicJob.ScheduleFunc(periodicJob.nextRunAt) if periodicJob.ID != "" { + periodicJobsToKeepAlive[handle] = periodicJob periodicJobUpsertParams.Jobs = append(periodicJobUpsertParams.Jobs, &riverpilot.PeriodicJobUpsertParams{ ID: periodicJob.ID, NextRunAt: periodicJob.nextRunAt, @@ -488,7 +501,7 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { } }() - s.insertBatch(ctx, insertParamsMany, periodicJobUpsertParams) + s.insertBatch(ctx, insertParamsMany, periodicJobUpsertParams, periodicJobsToKeepAlive) case <-s.recalculateNextRun: if !timerUntilNextRun.Stop() { @@ -518,7 +531,7 @@ func (s *PeriodicJobEnqueuer) Start(ctx context.Context) error { return nil } -func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany []*rivertype.JobInsertParams, periodicJobUpsertParams *riverpilot.PeriodicJobUpsertManyParams) { +func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany []*rivertype.JobInsertParams, periodicJobUpsertParams *riverpilot.PeriodicJobUpsertManyParams, periodicJobsToKeepAlive map[rivertype.PeriodicJobHandle]*PeriodicJob) { if len(insertParamsMany) < 1 && len(periodicJobUpsertParams.Jobs) < 1 { return } @@ -554,6 +567,16 @@ func (s *PeriodicJobEnqueuer) insertBatch(ctx context.Context, insertParamsMany return } + if len(periodicJobsToKeepAlive) > 0 { + s.mu.Lock() + for handle, periodicJob := range periodicJobsToKeepAlive { + if s.periodicJobs[handle] == periodicJob { + periodicJob.keepAlive = true + } + } + s.mu.Unlock() + } + if len(insertParamsMany) > 0 { s.TestSignals.InsertedJobs.Signal(struct{}{}) } @@ -614,8 +637,16 @@ func (s *PeriodicJobEnqueuer) periodicJobKeepAliveAndReapPeriodically(ctx contex defer s.mu.RUnlock() if len(s.periodicJobIDs) > 0 { + // Can be empty. The call still reaps expired rows. + ids := make([]string, 0, len(s.periodicJobIDs)) + for id, handle := range s.periodicJobIDs { + if s.periodicJobs[handle].keepAlive { + ids = append(ids, id) + } + } + if _, err := s.Config.Pilot.PeriodicJobKeepAliveAndReap(ctx, s.exec, &riverpilot.PeriodicJobKeepAliveAndReapParams{ - ID: maputil.Keys(s.periodicJobIDs), + ID: ids, Schema: s.Config.Schema, }); err != nil { s.Logger.ErrorContext(ctx, s.Name+": Error executing periodic job keep alive and reap", "error", err.Error()) diff --git a/internal/maintenance/periodic_job_enqueuer_keep_alive_test.go b/internal/maintenance/periodic_job_enqueuer_keep_alive_test.go new file mode 100644 index 000000000..22e68f79a --- /dev/null +++ b/internal/maintenance/periodic_job_enqueuer_keep_alive_test.go @@ -0,0 +1,260 @@ +package maintenance + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/riverqueue/river/riverdriver" + "github.com/riverqueue/river/rivershared/riverpilot" + "github.com/riverqueue/river/rivershared/riversharedtest" + "github.com/riverqueue/river/rivershared/testsignal" + "github.com/riverqueue/river/rivertype" +) + +type periodicKeepAliveExecutor struct { + riverdriver.Executor + + beginErr error + commitBarrier <-chan struct{} + failCommit atomic.Bool + testSignals periodicKeepAliveExecutorTestSignals +} + +func (e *periodicKeepAliveExecutor) Begin(context.Context) (riverdriver.ExecutorTx, error) { + if e.beginErr != nil { + return nil, e.beginErr + } + return &periodicKeepAliveTx{executor: e}, nil +} + +type periodicKeepAliveExecutorTestSignals struct { + Commit testsignal.TestSignal[error] + CommitStarted testsignal.TestSignal[struct{}] +} + +func (s *periodicKeepAliveExecutorTestSignals) Init(tb testing.TB) { + tb.Helper() + + s.Commit.Init(tb) + s.CommitStarted.Init(tb) +} + +type periodicKeepAliveTx struct { + riverdriver.ExecutorTx + + executor *periodicKeepAliveExecutor +} + +func (tx *periodicKeepAliveTx) Commit(ctx context.Context) error { + tx.executor.testSignals.CommitStarted.Signal(struct{}{}) + if tx.executor.commitBarrier != nil { + select { + case <-tx.executor.commitBarrier: + case <-ctx.Done(): + } + } + err := ctx.Err() + if err == nil && tx.executor.failCommit.Load() { + err = errors.New("commit failed") + } + tx.executor.testSignals.Commit.Signal(err) + return err +} + +func (*periodicKeepAliveTx) Rollback(context.Context) error { return nil } + +func TestPeriodicJobEnqueuerKeepAliveAfterPersistence(t *testing.T) { + t.Parallel() + + type testBundle struct { + blockSchedule bool + executor *periodicKeepAliveExecutor + initialRows []*riverpilot.PeriodicJob + keepAliveIDs testsignal.TestSignal[[]string] + now time.Time + pendingScheduleFunc func(time.Time) time.Time + releaseSchedule func() + scheduleEntered testsignal.TestSignal[struct{}] + service *PeriodicJobEnqueuer + upsertErr error + } + + setup := func(t *testing.T) (*testBundle, context.Context) { + t.Helper() + + ctx, cancel := context.WithCancel(context.Background()) + bundle := &testBundle{executor: &periodicKeepAliveExecutor{}, now: time.Now().UTC()} + bundle.executor.testSignals.Init(t) + bundle.keepAliveIDs.Init(t) + bundle.scheduleEntered.Init(t) + bundle.initialRows = []*riverpilot.PeriodicJob{{ID: "fresh", NextRunAt: bundle.now.Add(2 * time.Hour)}} + + // Hold a schedule callback while the actual keep-alive loop runs. + scheduleBarrier := make(chan struct{}) + var releaseOnce sync.Once + bundle.releaseSchedule = func() { releaseOnce.Do(func() { close(scheduleBarrier) }) } + bundle.pendingScheduleFunc = func(now time.Time) time.Time { + if bundle.blockSchedule { + bundle.scheduleEntered.Signal(struct{}{}) + <-scheduleBarrier + } + return now.Add(time.Hour) + } + + pilot := NewPilotPeriodicJobMock() + pilot.PeriodicJobGetAllMock = func(context.Context, riverdriver.Executor, *riverpilot.PeriodicJobGetAllParams) ([]*riverpilot.PeriodicJob, error) { + return bundle.initialRows, nil + } + pilot.PeriodicJobKeepAliveAndReapMock = func(_ context.Context, _ riverdriver.Executor, params *riverpilot.PeriodicJobKeepAliveAndReapParams) ([]*riverpilot.PeriodicJob, error) { + bundle.keepAliveIDs.Signal(params.ID) + return nil, nil + } + pilot.PeriodicJobUpsertManyMock = func(context.Context, riverdriver.Executor, *riverpilot.PeriodicJobUpsertManyParams) ([]*riverpilot.PeriodicJob, error) { + return nil, bundle.upsertErr + } + constructor := func() (*rivertype.JobInsertParams, error) { return nil, ErrNoJobToInsert } + archetype := riversharedtest.BaseServiceArchetype(t) + archetype.Time.StubNow(bundle.now) + var err error + bundle.service, err = NewPeriodicJobEnqueuer(archetype, &PeriodicJobEnqueuerConfig{ + PeriodicJobs: []*PeriodicJob{ + {ID: "fresh", ConstructorFunc: constructor, ScheduleFunc: func(now time.Time) time.Time { return now.Add(2 * time.Hour) }}, + {ID: "pending", ConstructorFunc: constructor, ScheduleFunc: bundle.pendingScheduleFunc}, + }, + Pilot: pilot, + }, bundle.executor) + require.NoError(t, err) + bundle.service.StaggerStartupDisable(true) + bundle.service.TestSignals.Init(t) + t.Cleanup(func() { + cancel() + bundle.releaseSchedule() + bundle.service.Stop() + }) + return bundle, ctx + } + + runKeepAlive := func(ctx context.Context, t *testing.T, bundle *testBundle) []string { + t.Helper() + + ctx, cancel := context.WithCancel(ctx) + var stopped testsignal.TestSignal[struct{}] + stopped.Init(t) + require.NoError(t, bundle.service.periodicJobKeepAliveAndReapPeriodically(ctx, true, func() {}, func() { stopped.Signal(struct{}{}) })) + defer func() { cancel(); stopped.WaitOrTimeout() }() + return bundle.keepAliveIDs.WaitOrTimeout() + } + + t.Run("InitialPersistence", func(t *testing.T) { + t.Parallel() + + for _, outcome := range []string{"BeginFailed", "CommitFailed", "Committed", "UpsertFailed"} { + t.Run(outcome, func(t *testing.T) { + t.Parallel() + + bundle, ctx := setup(t) + switch outcome { + case "BeginFailed": + bundle.executor.beginErr = errors.New("begin failed") + case "CommitFailed": + bundle.executor.failCommit.Store(true) + case "UpsertFailed": + bundle.upsertErr = errors.New("upsert failed") + } + require.NoError(t, bundle.service.Start(ctx)) + bundle.service.TestSignals.EnteredLoop.WaitOrTimeout() + bundle.service.TestSignals.PeriodicJobKeepAliveAndReap.WaitOrTimeout() + bundle.keepAliveIDs.WaitOrTimeout() + + expected := []string{"fresh"} + if outcome == "Committed" { + expected = append(expected, "pending") + } + require.ElementsMatch(t, expected, runKeepAlive(ctx, t, bundle)) + }) + } + }) + + t.Run("LaterCommittedOccurrence", func(t *testing.T) { + t.Parallel() + + bundle, ctx := setup(t) + bundle.executor.failCommit.Store(true) + require.NoError(t, bundle.service.Start(ctx)) + bundle.service.TestSignals.EnteredLoop.WaitOrTimeout() + bundle.service.TestSignals.PeriodicJobKeepAliveAndReap.WaitOrTimeout() + bundle.keepAliveIDs.WaitOrTimeout() + + bundle.executor.failCommit.Store(false) + bundle.service.Time.StubNow(bundle.now.Add(time.Hour)) + bundle.service.recalculateNextRun <- struct{}{} + bundle.service.TestSignals.PeriodicJobUpserted.WaitOrTimeout() + + require.ElementsMatch(t, []string{"fresh", "pending"}, runKeepAlive(ctx, t, bundle)) + }) + + t.Run("PendingStartup", func(t *testing.T) { + t.Parallel() + + bundle, ctx := setup(t) + bundle.blockSchedule = true + require.NoError(t, bundle.service.Start(ctx)) + bundle.scheduleEntered.WaitOrTimeout() + + require.ElementsMatch(t, []string{"fresh"}, bundle.keepAliveIDs.WaitOrTimeout()) + }) + + t.Run("RemovedAndReadded", func(t *testing.T) { + t.Parallel() + + bundle, ctx := setup(t) + // Replace the registration while its original transaction is committing. + commitBarrier := make(chan struct{}) + var releaseOnce sync.Once + releaseCommit := func() { releaseOnce.Do(func() { close(commitBarrier) }) } + t.Cleanup(releaseCommit) + bundle.executor.commitBarrier = commitBarrier + require.NoError(t, bundle.service.Start(ctx)) + bundle.executor.testSignals.CommitStarted.WaitOrTimeout() + bundle.service.mu.RLock() + pendingJob := bundle.service.periodicJobs[bundle.service.periodicJobIDs["pending"]] + bundle.service.mu.RUnlock() + require.True(t, bundle.service.RemoveByID("pending")) + bundle.blockSchedule = true + // Reuse the pointer so a late commit could mark the new registration. + _, err := bundle.service.AddSafely(pendingJob) + require.NoError(t, err) + releaseCommit() + bundle.scheduleEntered.WaitOrTimeout() + bundle.service.TestSignals.PeriodicJobKeepAliveAndReap.WaitOrTimeout() + bundle.keepAliveIDs.WaitOrTimeout() + + require.ElementsMatch(t, []string{"fresh"}, runKeepAlive(ctx, t, bundle)) + }) + + t.Run("Restarted", func(t *testing.T) { + t.Parallel() + + bundle, ctx := setup(t) + require.NoError(t, bundle.service.Start(ctx)) + bundle.service.TestSignals.EnteredLoop.WaitOrTimeout() + bundle.service.TestSignals.PeriodicJobKeepAliveAndReap.WaitOrTimeout() + bundle.keepAliveIDs.WaitOrTimeout() + bundle.service.Stop() + + bundle.initialRows = nil + bundle.blockSchedule = true + bundle.service.TestSignals.Init(t) + require.NoError(t, bundle.service.Start(ctx)) + bundle.scheduleEntered.WaitOrTimeout() + bundle.service.TestSignals.PeriodicJobKeepAliveAndReap.WaitOrTimeout() + + require.Empty(t, bundle.keepAliveIDs.WaitOrTimeout()) + }) +} diff --git a/internal/maintenance/periodic_job_enqueuer_test.go b/internal/maintenance/periodic_job_enqueuer_test.go index 28c1be8e3..bb38d153a 100644 --- a/internal/maintenance/periodic_job_enqueuer_test.go +++ b/internal/maintenance/periodic_job_enqueuer_test.go @@ -845,7 +845,11 @@ func TestPeriodicJobEnqueuer(t *testing.T) { var periodicJobKeepAliveAndReapMockCalled bool bundle.pilotMock.PeriodicJobKeepAliveAndReapMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobKeepAliveAndReapParams) ([]*riverpilot.PeriodicJob, error) { periodicJobKeepAliveAndReapMockCalled = true - require.ElementsMatch(t, []string{"pilot_startup_new_job", "pilot_startup_durable_1h", "pilot_startup_durable_2h"}, params.ID) + // The first keep-alive tick can run before or after the startup + // upsert commits, so the new job may or may not be included. + require.Subset(t, []string{"pilot_startup_new_job", "pilot_startup_durable_1h", "pilot_startup_durable_2h"}, params.ID) + require.Contains(t, params.ID, "pilot_startup_durable_1h") + require.Contains(t, params.ID, "pilot_startup_durable_2h") require.Equal(t, bundle.schema, params.Schema) return nil, nil } @@ -915,11 +919,64 @@ func TestPeriodicJobEnqueuer(t *testing.T) { UpdatedAt: now, }}, Schema: bundle.schema, - }) + }, nil) require.False(t, periodicJobUpsertManyMockCalled) }) + // Runs a durable batch through a real database transaction. A job becomes + // eligible for keep-alive only when its insertion and next run time + // commit together; a rolled back batch leaves no job and no eligibility. + t.Run("InsertBatchKeepsAliveOnlyAfterCommit", func(t *testing.T) { + t.Parallel() + + for _, committed := range []bool{false, true} { + t.Run(fmt.Sprintf("Committed=%t", committed), func(t *testing.T) { + t.Parallel() + + svc, bundle := setup(t) + + bundle.pilotMock.PeriodicJobUpsertManyMock = func(ctx context.Context, exec riverdriver.Executor, params *riverpilot.PeriodicJobUpsertManyParams) ([]*riverpilot.PeriodicJob, error) { + if !committed { + return nil, errors.New("upsert failed after job insert") + } + return nil, nil + } + + periodicJob := &PeriodicJob{ + ID: "insert_batch_keep_alive", + ConstructorFunc: jobConstructorFunc("insert_batch_keep_alive", false), + ScheduleFunc: periodicIntervalSchedule(time.Hour), + } + handle, err := svc.AddSafely(periodicJob) + require.NoError(t, err) + + insertParams, err := periodicJob.ConstructorFunc() + require.NoError(t, err) + + now := time.Now().UTC() + svc.insertBatch(ctx, []*rivertype.JobInsertParams{insertParams}, &riverpilot.PeriodicJobUpsertManyParams{ + Jobs: []*riverpilot.PeriodicJobUpsertParams{{ + ID: periodicJob.ID, + NextRunAt: now.Add(time.Hour), + UpdatedAt: now, + }}, + Schema: bundle.schema, + }, map[rivertype.PeriodicJobHandle]*PeriodicJob{handle: periodicJob}) + + expectedNumJobs := 0 + if committed { + expectedNumJobs = 1 + } + requireNJobs(t, bundle, "insert_batch_keep_alive", expectedNumJobs) + + svc.mu.RLock() + defer svc.mu.RUnlock() + require.Equal(t, committed, periodicJob.keepAlive) + }) + } + }) + t.Run("InvokesPilotDueDurableJobRunsOnce", func(t *testing.T) { t.Parallel() diff --git a/js/CHANGELOG.md b/js/CHANGELOG.md index c1a878de1..99db0142d 100644 --- a/js/CHANGELOG.md +++ b/js/CHANGELOG.md @@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Refresh a durable periodic job's saved row only after its next run time is loaded at startup or committed. When a startup write rolled back, keep-alive could revive an expired row, and the next leader enqueued every run since the row's old next run time. [PR #1540](https://github.com/riverqueue/river/pull/1540). - Fail retries that omit a saved cursor-only step, while preserving their checkpoints for recovery. Jobs with an existing step marker retain their behavior, including legacy stale cursor entries. [PR #1534](https://github.com/riverqueue/river/pull/1534). - Replace whole top-level metadata values on SQLite, including JSON null, matching Postgres. Completed resumable cursor steps no longer run again because of stale cursor entries. [PR #1533](https://github.com/riverqueue/river/pull/1533). diff --git a/js/etc/riverqueue.unstable-driver.api.md b/js/etc/riverqueue.unstable-driver.api.md index abb95301a..9830babf3 100644 --- a/js/etc/riverqueue.unstable-driver.api.md +++ b/js/etc/riverqueue.unstable-driver.api.md @@ -1154,22 +1154,21 @@ export interface PeerOutcome { ```ts /** - * A pilot's durable storage for periodic job schedules, so the next run of - * a periodic job with an `id` survives leader changes and restarts. It - * mirrors River for Go's periodic-job pilot operations. + * A pilot's durable storage for periodic schedules. A job with an `id` keeps + * its next run time across leader changes and restarts. * - * The leader calls `getAll` when it starts enqueuing periodic jobs and seeds - * each job's next run from the matching record; calls `upsertMany` inside the - * same transaction that inserts each batch of periodic jobs; and calls - * `keepAliveAndReap` with the registered IDs every ten minutes so the store - * can delete records for jobs no client registers anymore. + * At startup, the leader loads saved run times with `getAll`. It writes new + * times through `upsertMany` in the same transaction as periodic insertions. + * Every ten minutes, the leader calls `keepAliveAndReap` with the IDs whose + * run times were loaded at startup or committed since. The store refreshes + * those records and reaps expired ones, even when the list is empty. */ export interface PeriodicJobStore { - /** Return every durable periodic job record. */ + /** Return saved schedules that have not expired. */ getAll(options: { readonly signal: AbortSignal; }): PromiseLike; - /** Refresh records for `ids` and delete records not refreshed recently. */ + /** Refresh records for `ids` and reap expired records, even when `ids` is empty. */ keepAliveAndReap( ids: readonly string[], options: { diff --git a/js/src/periodic-job-store.ts b/js/src/periodic-job-store.ts index 40ed3d41f..70ab7f50d 100644 --- a/js/src/periodic-job-store.ts +++ b/js/src/periodic-job-store.ts @@ -8,22 +8,21 @@ export interface DurablePeriodicJobUpsert { } /** - * A pilot's durable storage for periodic job schedules, so the next run of - * a periodic job with an `id` survives leader changes and restarts. It - * mirrors River for Go's periodic-job pilot operations. + * A pilot's durable storage for periodic schedules. A job with an `id` keeps + * its next run time across leader changes and restarts. * - * The leader calls `getAll` when it starts enqueuing periodic jobs and seeds - * each job's next run from the matching record; calls `upsertMany` inside the - * same transaction that inserts each batch of periodic jobs; and calls - * `keepAliveAndReap` with the registered IDs every ten minutes so the store - * can delete records for jobs no client registers anymore. + * At startup, the leader loads saved run times with `getAll`. It writes new + * times through `upsertMany` in the same transaction as periodic insertions. + * Every ten minutes, the leader calls `keepAliveAndReap` with the IDs whose + * run times were loaded at startup or committed since. The store refreshes + * those records and reaps expired ones, even when the list is empty. */ export interface PeriodicJobStore { - /** Return every durable periodic job record. */ + /** Return saved schedules that have not expired. */ getAll(options: { readonly signal: AbortSignal; }): PromiseLike; - /** Refresh records for `ids` and delete records not refreshed recently. */ + /** Refresh records for `ids` and reap expired records, even when `ids` is empty. */ keepAliveAndReap( ids: readonly string[], options: { readonly signal: AbortSignal } diff --git a/js/src/periodic-persistence.test.ts b/js/src/periodic-persistence.test.ts new file mode 100644 index 000000000..69fae22f2 --- /dev/null +++ b/js/src/periodic-persistence.test.ts @@ -0,0 +1,120 @@ +import { describe, expect, it } from "vitest"; + +import { defineJob } from "./job-definition.js"; +import { + advancePeriodicJobs, + periodicJob, + periodicJobKeepAliveIds, + PeriodicJobs, + resetPeriodicJobs, +} from "./periodic.js"; + +const report = defineJob<{ scope: string }>()({ kind: "periodic_report" }); +const start = Temporal.Instant.from("2026-08-30T12:00:00Z"); + +function hourly(id: string) { + return periodicJob({ + args: { scope: id }, + every: { hours: 1 }, + id, + job: report, + }); +} + +function advance(jobs: PeriodicJobs, now = start) { + return advancePeriodicJobs(jobs, now, new Map(), (_job, error) => { + throw error; + }); +} + +describe("periodic registration persistence", () => { + it("keeps only saved registrations eligible before a batch commits", () => { + const jobs = new PeriodicJobs([hourly("fresh"), hourly("pending")]); + const saved = new Map([["fresh", start.add({ hours: 2 })]]); + resetPeriodicJobs(jobs, saved); + advancePeriodicJobs(jobs, start, saved, (_job, error) => { + throw error; + }); + + expect(periodicJobKeepAliveIds(jobs)).toEqual(["fresh"]); + }); + + it("ignores a commit for a removed registration", () => { + const jobs = new PeriodicJobs(); + const handle = jobs.add(hourly("hourly")); + const oldBatch = advance(jobs); + jobs.remove(handle); + jobs.add(hourly("hourly")); + oldBatch.markPersisted(); + + expect(periodicJobKeepAliveIds(jobs)).toEqual([]); + advance(jobs).markPersisted(); + expect(periodicJobKeepAliveIds(jobs)).toEqual(["hourly"]); + }); + + it("ignores a commit from an earlier leadership term", () => { + const jobs = new PeriodicJobs([hourly("hourly")]); + const oldBatch = advance(jobs); + resetPeriodicJobs(jobs); + oldBatch.markPersisted(); + + expect(periodicJobKeepAliveIds(jobs)).toEqual([]); + const newBatch = advance(jobs); + newBatch.markPersisted(); + expect(periodicJobKeepAliveIds(jobs)).toEqual(["hourly"]); + }); + + it("keeps a stopped schedule pending until it has a saved position", () => { + const jobs = new PeriodicJobs([ + periodicJob({ + args: { scope: "stopped" }, + id: "stopped", + job: report, + schedule: { next: () => null }, + }), + ]); + const batch = advance(jobs); + expect(batch.durableUpdates).toEqual([]); + batch.markPersisted(); + + expect(periodicJobKeepAliveIds(jobs)).toEqual([]); + }); + + it("promotes only registrations included in a committed batch", () => { + const jobs = new PeriodicJobs([hourly("first")]); + const firstBatch = advance(jobs); + jobs.add(hourly("second")); + const secondBatch = advance(jobs); + firstBatch.markPersisted(); + + expect(periodicJobKeepAliveIds(jobs)).toEqual(["first"]); + secondBatch.markPersisted(); + expect(periodicJobKeepAliveIds(jobs)).toEqual(["first", "second"]); + }); + + it("marks a registration after a later occurrence commits", () => { + const jobs = new PeriodicJobs([hourly("hourly")]); + advance(jobs); + advance(jobs, start.add({ hours: 1 })).markPersisted(); + + expect(periodicJobKeepAliveIds(jobs)).toEqual(["hourly"]); + }); + + it("clears eligibility when a new leadership term starts", () => { + const jobs = new PeriodicJobs([hourly("hourly")]); + advance(jobs).markPersisted(); + resetPeriodicJobs(jobs); + + expect(periodicJobKeepAliveIds(jobs)).toEqual([]); + resetPeriodicJobs(jobs, new Map([["hourly", start.add({ hours: 1 })]])); + expect(periodicJobKeepAliveIds(jobs)).toEqual(["hourly"]); + }); + + it("preserves eligibility after a later failed occurrence", () => { + const jobs = new PeriodicJobs([hourly("hourly")]); + advance(jobs).markPersisted(); + advance(jobs, start.add({ hours: 1 })); + + expect(periodicJobKeepAliveIds(jobs)).toEqual(["hourly"]); + }); +}); diff --git a/js/src/periodic.ts b/js/src/periodic.ts index 45cfa6027..d5508d1c3 100644 --- a/js/src/periodic.ts +++ b/js/src/periodic.ts @@ -175,9 +175,11 @@ export interface PeriodicJobsStartParams { } interface PeriodicEntry { + generation: number; readonly handle: number; initialized: boolean; readonly job: PeriodicJob; + keepAlive: boolean; nextRun: Temporal.Instant | null; } @@ -194,6 +196,8 @@ export interface DuePeriodicBatch { readonly id: string; readonly nextRunAt: Temporal.Instant; }[]; + /** Call after this batch's transaction commits, so keep-alive can refresh its registrations. */ + markPersisted(): void; readonly occurrences: readonly DuePeriodicOccurrence[]; } @@ -273,9 +277,11 @@ export class PeriodicJobs { const handles = jobs.map((job) => { const value = this.#nextHandle++; this.#entries.set(value, { + generation: 0, handle: value, initialized: false, job, + keepAlive: false, nextRun: null, }); return Object.freeze({ "~periodicJobHandle": value }); @@ -355,14 +361,29 @@ export function setPeriodicJobsChangeHandler( registryInternals(registry).setChangeHandler(change); } -/** @internal Forget all scheduling state when leadership starts or ends. */ -export function resetPeriodicJobs(registry: PeriodicJobs): void { +/** @internal Reset scheduling state. Only registrations with a saved next run time are eligible for keep-alive. */ +export function resetPeriodicJobs( + registry: PeriodicJobs, + durableNextRuns?: ReadonlyMap +): void { for (const entry of registryInternals(registry).entries.values()) { + entry.generation += 1; entry.initialized = false; + entry.keepAlive = + entry.job.id !== null && durableNextRuns?.has(entry.job.id) === true; entry.nextRun = null; } } +/** @internal IDs whose saved schedules keep-alive may refresh. */ +export function periodicJobKeepAliveIds( + registry: PeriodicJobs +): readonly string[] { + return [...registryInternals(registry).entries.values()].flatMap( + ({ job, keepAlive }) => (job.id !== null && keepAlive ? [job.id] : []) + ); +} + /** @internal IDs of registered periodic jobs, for durable keep-alive. */ export function periodicJobIds(registry: PeriodicJobs): readonly string[] { return [...registryInternals(registry).entries.values()].flatMap(({ job }) => @@ -410,6 +431,7 @@ export function advancePeriodicJobs( ): DuePeriodicBatch { const occurrences: DuePeriodicOccurrence[] = []; const durableUpdates: { id: string; nextRunAt: Temporal.Instant }[] = []; + const entriesToPersist = new Map(); const dueBefore = now.add(DUE_MARGIN); const entries = [...registryInternals(registry).entries.values()].sort( (left, right) => left.handle - right.handle @@ -424,6 +446,7 @@ export function advancePeriodicJobs( seeded ?? safeNextOccurrence(entry.job, now, onScheduleError); if (id !== null && entry.nextRun !== null) { durableUpdates.push({ id, nextRunAt: entry.nextRun }); + entriesToPersist.set(entry, entry.generation); } if (entry.job.runOnStart) { occurrences.push({ job: entry.job, scheduledAt: now }); @@ -444,9 +467,20 @@ export function advancePeriodicJobs( ); if (id !== null && entry.nextRun !== null) { durableUpdates.push({ id, nextRunAt: entry.nextRun }); + entriesToPersist.set(entry, entry.generation); } } - return { durableUpdates, occurrences }; + return { + durableUpdates, + markPersisted: () => { + for (const [entry, generation] of entriesToPersist) { + // Marking a removed entry is harmless because `add` never reuses one. + // The generation check rejects commits from an earlier leadership term. + if (entry.generation === generation) entry.keepAlive = true; + } + }, + occurrences, + }; } /** diff --git a/js/src/services.test.ts b/js/src/services.test.ts index 5a85bd7f2..7c199d9c7 100644 --- a/js/src/services.test.ts +++ b/js/src/services.test.ts @@ -206,6 +206,98 @@ describe("RuntimeServices", () => { ]); }); + it.each([ + { outcome: "commit failure", saved: true }, + { outcome: "committed", saved: true }, + { outcome: "upsert failure", saved: false }, + { outcome: "upsert failure", saved: true }, + ] as const)( + "keeps only saved periodic registrations alive after $outcome (saved row: $saved)", + async ({ outcome, saved }) => { + const shared = new SharedLeadership(); + const definition = defineJob<{ n: number }>()({ kind: "periodic" }); + const now = Temporal.Instant.from("2026-08-30T12:00:00Z"); + const kept: (readonly string[])[] = []; + const driver: RuntimeDriver = { + ...serviceDriver(shared), + operationScope: async ( + _tx: unknown, + callback: (tx: unknown) => Promise + ) => { + const result = await callback("tx"); + if (outcome === "commit failure") throw new Error("commit failed"); + return result; + }, + }; + const periodicJobs = new PeriodicJobs([ + periodicJob({ + args: { n: 1 }, + every: { hours: 2 }, + id: "fresh", + job: definition, + }), + periodicJob({ + args: { n: 2 }, + every: { hours: 1 }, + id: "pending", + job: definition, + }), + ]); + const services = new RuntimeServices({ + client: { insertMany: () => Promise.resolve([]) } as unknown as Client, + clientId: "leader", + driver, + emit: () => Promise.resolve(), + logger: { error: () => {}, warn: () => {} }, + maintenance: { + electionIntervalMs: 5, + jobCleanerIntervalMs: 60_000, + queueCleanerIntervalMs: 60_000, + rescuerIntervalMs: 60_000, + schedulerIntervalMs: 60_000, + }, + now: () => now, + onPeriodicJobsStart: () => Promise.resolve(), + periodicJobStore: { + getAll: () => + Promise.resolve( + saved + ? [ + { + createdAt: now, + id: "fresh", + nextRunAt: now.add({ hours: 2 }), + updatedAt: now, + }, + ] + : [] + ), + keepAliveAndReap: (ids) => { + kept.push(ids); + return Promise.resolve(); + }, + upsertMany: () => + outcome === "upsert failure" + ? Promise.reject(new Error("upsert failed")) + : Promise.resolve(), + }, + periodicJobs, + rescue: () => Promise.resolve(null), + }); + const controller = new AbortController(); + const running = services.run(controller.signal); + try { + await waitUntil(() => kept.length > 0); + } finally { + controller.abort(); + await running; + } + expect(kept).toEqual([ + outcome === "committed" ? ["fresh", "pending"] : saved ? ["fresh"] : [], + ]); + } + ); + it("resigns its own term once maintenance fails to start three times, like Go", async () => { const shared = new SharedLeadership(); const base = serviceDriver(shared); diff --git a/js/src/services.ts b/js/src/services.ts index ff798b4a4..803d3bc95 100644 --- a/js/src/services.ts +++ b/js/src/services.ts @@ -15,6 +15,7 @@ import { buildPeriodicInsert, nextPeriodicRunAt, periodicJobIds, + periodicJobKeepAliveIds, resetPeriodicJobs, setPeriodicJobsChangeHandler, } from "./periodic.js"; @@ -758,8 +759,11 @@ export class RuntimeServices { durableJobs: Object.freeze([...durableJobs]), periodicJobs: this.#periodicJobs, }); - resetPeriodicJobs(this.#periodicJobs); - return new Map(durableJobs.map(({ id, nextRunAt }) => [id, nextRunAt])); + const nextRuns = new Map( + durableJobs.map(({ id, nextRunAt }) => [id, nextRunAt]) + ); + resetPeriodicJobs(this.#periodicJobs, nextRuns); + return nextRuns; } catch (error: unknown) { if (signal.aborted) return null; this.#lastError = errorMessage(error); @@ -827,6 +831,9 @@ export class RuntimeServices { if (items.length > 0) await this.#client.insertMany(items, { tx }); await store.upsertMany(tx, upserts); }); + if (!signal.aborted && this.#ownsLeadership(term)) { + batch.markPersisted(); + } } else if (items.length > 0) { await this.#client.insertMany(items); } @@ -843,8 +850,8 @@ export class RuntimeServices { async #keepPeriodicJobsAlive(signal: AbortSignal): Promise { const store = this.#periodicJobStore; if (store === undefined) return; - const ids = periodicJobIds(this.#periodicJobs); - if (ids.length === 0) return; + if (periodicJobIds(this.#periodicJobs).length === 0) return; + const ids = periodicJobKeepAliveIds(this.#periodicJobs); try { await store.keepAliveAndReap(ids, { signal }); } catch (error: unknown) {