fix: remove jobs and respect stopTime in NextRuns when WithStopDateTime is set (#922)

Author: jcheyer

job.go

@@ -1635,7 +1635,11 @@ func (j job) NextRuns(count int) ([]time.Time, error) {
 		}
 
 		from := out[i-1]
-		out[i] = ij.next(from)
+		next := ij.next(from)
+		if !ij.stopTime.IsZero() && !next.Before(ij.stopTime) {
+			return out[:i], nil
+		}
+		out[i] = next
 	}
 
 	return out, nil

job_test.go

@@ -1043,6 +1043,37 @@ func TestJob_NextRuns(t *testing.T) {
 	}
 }
 
+func TestJob_NextRuns_StopTime(t *testing.T) {
+	stopTime := time.Now().Add(350 * time.Millisecond)
+
+	s := newTestScheduler(t)
+	j, err := s.NewJob(
+		DurationJob(100*time.Millisecond),
+		NewTask(func() {}),
+		WithStopAt(WithStopDateTime(stopTime)),
+	)
+	require.NoError(t, err)
+
+	s.Start()
+	time.Sleep(50 * time.Millisecond)
+
+	// nextScheduled must not contain any time at or after stopTime
+	ij := requestJob(j.ID(), j.(*job).jobOutRequest)
+	require.NotNil(t, ij)
+	for _, ns := range ij.nextScheduled {
+		assert.True(t, ns.Before(stopTime), "nextScheduled contains time after stopTime: %v >= %v", ns, stopTime)
+	}
+
+	// NextRuns with a large count must all be before stopTime
+	runs, err := j.NextRuns(100)
+	require.NoError(t, err)
+	for _, r := range runs {
+		assert.True(t, r.Before(stopTime), "NextRuns returned time after stopTime: %v >= %v", r, stopTime)
+	}
+
+	require.NoError(t, s.Shutdown())
+}
+
 func TestJob_PanicOccurred(t *testing.T) {
 	gotCh := make(chan any)
 	errCh := make(chan error)

scheduler.go

@@ -363,6 +363,7 @@ func (s *scheduler) selectExecJobsOutForRescheduling(id uuid.UUID) {
 	}
 
 	if j.stopTimeReached(s.now()) {
+		s.selectRemoveJob(id)
 		return
 	}
 
@@ -422,6 +423,11 @@ func (s *scheduler) selectExecJobsOutForRescheduling(id uuid.UUID) {
 		}
 	}
 
+	if !j.stopTime.IsZero() && !next.Before(j.stopTime) {
+		s.selectRemoveJob(id)
+		return
+	}
+
 	// Clean up any existing timer to prevent leaks
 	if j.timer != nil {
 		j.timer.Stop()
@@ -548,6 +554,13 @@ func (s *scheduler) selectNewJob(in newJobIn) {
 				}
 			}
 
+			if !j.stopTime.IsZero() && !next.Before(j.stopTime) {
+				s.jobs[j.id] = j
+				in.cancel()
+				s.selectRemoveJob(j.id)
+				return
+			}
+
 			id := j.id
 			j.timer = s.exec.clock.AfterFunc(next.Sub(s.now()), func() {
 				select {
@@ -609,6 +622,11 @@ func (s *scheduler) selectStart() {
 				}
 			}
 
+			if !j.stopTime.IsZero() && !next.Before(j.stopTime) {
+				s.selectRemoveJob(id)
+				continue
+			}
+
 			jobID := id
 			j.timer = s.exec.clock.AfterFunc(next.Sub(s.now()), func() {
 				select {

scheduler_test.go

@@ -3190,3 +3190,76 @@ func TestScheduler_JobSchedule(t *testing.T) {
 		})
 	}
 }
+
+func TestScheduler_WithStopDateTime_JobRemovedAfterStopTime(t *testing.T) {
+	defer verifyNoGoroutineLeaks(t)
+
+	t.Run("job is removed from scheduler after stop time elapses", func(t *testing.T) {
+		monitor := newTestSchedulerMonitor()
+		s := newTestScheduler(t, WithSchedulerMonitor(monitor))
+
+		_, err := s.NewJob(
+			DurationJob(50*time.Millisecond),
+			NewTask(func() {}),
+			WithStopAt(WithStopDateTime(time.Now().Add(200*time.Millisecond))),
+			WithStartAt(WithStartImmediately()),
+		)
+		require.NoError(t, err)
+
+		s.Start()
+
+		require.Eventually(t, func() bool {
+			return len(s.Jobs()) == 0
+		}, time.Second, 10*time.Millisecond, "job should be removed after stop time")
+
+		assert.GreaterOrEqual(t, monitor.getJobUnregCount(), int64(1), "monitor should receive JobUnregistered notification")
+
+		require.NoError(t, s.Shutdown())
+	})
+
+	t.Run("job added before start is removed on start when stop time already elapsed", func(t *testing.T) {
+		monitor := newTestSchedulerMonitor()
+		s := newTestScheduler(t, WithSchedulerMonitor(monitor))
+
+		_, err := s.NewJob(
+			DurationJob(time.Hour),
+			NewTask(func() {}),
+			WithStopAt(WithStopDateTime(time.Now().Add(100*time.Millisecond))),
+		)
+		require.NoError(t, err)
+
+		// wait until stop time has passed, then start the scheduler
+		time.Sleep(150 * time.Millisecond)
+		s.Start()
+
+		require.Eventually(t, func() bool {
+			return len(s.Jobs()) == 0
+		}, time.Second, 10*time.Millisecond, "job should be removed when scheduler starts after stop time")
+
+		require.NoError(t, s.Shutdown())
+	})
+
+	t.Run("RemoveJob on already auto-removed job returns ErrJobNotFound", func(t *testing.T) {
+		s := newTestScheduler(t)
+
+		j, err := s.NewJob(
+			DurationJob(50*time.Millisecond),
+			NewTask(func() {}),
+			WithStopAt(WithStopDateTime(time.Now().Add(150*time.Millisecond))),
+			WithStartAt(WithStartImmediately()),
+		)
+		require.NoError(t, err)
+
+		s.Start()
+
+		require.Eventually(t, func() bool {
+			return len(s.Jobs()) == 0
+		}, time.Second, 10*time.Millisecond, "job should be auto-removed after stop time")
+
+		// Explicitly removing an already auto-removed job should return ErrJobNotFound
+		err = s.RemoveJob(j.ID())
+		assert.ErrorIs(t, err, ErrJobNotFound)
+
+		require.NoError(t, s.Shutdown())
+	})
+}