Refactor cron lifecycle

dkron/agent.go

@@ -245,8 +245,7 @@ func (a *Agent) Stop() error {
 	a.logger.Info("agent: Called member stop, now stopping")
 
 	if a.config.Server && a.sched.Started() {
-		a.sched.Stop()
-		a.sched.ClearCron()
+		<-a.sched.Stop().Done()
 	}
 
 	if a.config.Server {

dkron/scheduler.go

@@ -1,6 +1,7 @@
 package dkron
 
 import (
+	"context"
 	"errors"
 	"expvar"
 	"strings"
@@ -35,7 +36,7 @@ type Scheduler struct {
 func NewScheduler(logger *logrus.Entry) *Scheduler {
 	schedulerStarted.Set(0)
 	return &Scheduler{
-		Cron:        nil,
+		Cron:        cron.New(cron.WithParser(extcron.NewParser())),
 		started:     false,
 		EntryJobMap: sync.Map{},
 		logger:      logger,
@@ -48,14 +49,10 @@ func (s *Scheduler) Start(jobs []*Job, agent *Agent) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	if s.Cron != nil {
-		// Stop the cron scheduler first and wait for all jobs to finish
-		s.Stop()
-		// Clear the cron scheduler
-		s.Cron = nil
+	if s.started {
+		return errors.New("scheduler: cron already started, should be stopped first")
 	}
-
-	s.Cron = cron.New(cron.WithParser(extcron.NewParser()))
+	s.ClearCron()
 
 	metrics.IncrCounter([]string{"scheduler", "start"}, 1)
 	for _, job := range jobs {
@@ -72,39 +69,37 @@ func (s *Scheduler) Start(jobs []*Job, agent *Agent) error {
 }
 
 // Stop stops the scheduler effectively not running any job.
-func (s *Scheduler) Stop() {
+func (s *Scheduler) Stop() context.Context {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	ctx := s.Cron.Stop()
 	if s.started {
 		s.logger.Debug("scheduler: Stopping scheduler")
-		<-s.Cron.Stop().Done()
 		s.started = false
-		// Keep Cron exists and let the jobs which have been scheduled can continue to finish,
-		// even the node's leadership will be revoked.
-		// Ignore the running jobs and make s.Cron to nil may cause whole process crashed.
-		//s.Cron = nil
 
 		// expvars
 		cronInspect.Do(func(kv expvar.KeyValue) {
 			kv.Value = nil
 		})
 	}
 	schedulerStarted.Set(0)
+	return ctx
 }
 
 // Restart the scheduler
 func (s *Scheduler) Restart(jobs []*Job, agent *Agent) {
 	s.Stop()
-	s.ClearCron()
 	if err := s.Start(jobs, agent); err != nil {
 		s.logger.Fatal(err)
 	}
 }
 
 // Clear cron separately, this can only be called when agent will be stop.
 func (s *Scheduler) ClearCron() {
-	s.Cron = nil
+	for _, e := range s.Cron.Entries() {
+		s.Cron.Remove(e.ID)
+	}
 }
 
 // Started will safely return if the scheduler is started or not

dkron/scheduler_test.go

@@ -1,9 +1,12 @@
 package dkron
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
+	"github.com/distribworks/dkron/v3/extcron"
+	"github.com/robfig/cron/v3"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -45,7 +48,7 @@ func TestSchedule(t *testing.T) {
 	assert.True(t, sched.Started())
 	assert.Len(t, sched.Cron.Entries(), 1)
 
-	sched.Cron.Remove(1)
+	sched.ClearCron()
 	assert.Len(t, sched.Cron.Entries(), 0)
 
 	sched.Stop()
@@ -69,3 +72,36 @@ func TestTimezoneAwareJob(t *testing.T) {
 	assert.Len(t, sched.Cron.Entries(), 1)
 	sched.Stop()
 }
+
+func TestScheduleStop(t *testing.T) {
+	log := getTestLogger()
+	sched := NewScheduler(log)
+
+	sched.Cron = cron.New(cron.WithParser(extcron.NewParser()))
+	sched.Cron.AddFunc("@every 2s", func() {
+		time.Sleep(time.Second * 5)
+		fmt.Println("function done")
+	})
+	sched.Cron.Start()
+	sched.started = true
+
+	testJob1 := &Job{
+		Name:           "cron_job",
+		Schedule:       "@every 2s",
+		Executor:       "shell",
+		ExecutorConfig: map[string]string{"command": "echo 'test1'", "shell": "true"},
+		Owner:          "John Dough",
+		OwnerEmail:     "[email protected]",
+	}
+	err := sched.Start([]*Job{testJob1}, &Agent{})
+	assert.Error(t, err)
+
+	// Wait for the job to start
+	time.Sleep(time.Second * 2)
+	<-sched.Stop().Done()
+	err = sched.Start([]*Job{testJob1}, &Agent{})
+	assert.NoError(t, err)
+
+	sched.Stop()
+	assert.False(t, sched.Started())
+}