feat(jobscheduler): add jobs after missed tick

Author: dbohdan

basis.go

@@ -41,6 +41,7 @@ const (
 	timestampFormat = "2006-01-02 15:04:05 -0700"
 
 	debounceInterval = 100 * time.Millisecond
+	maxMissedTime    = time.Hour
 	runInterval      = time.Second
 	scheduleInterval = time.Minute
 

jobconfig.go

@@ -33,14 +33,9 @@ func (j JobConfig) QueueName() string {
 	return j.Queue
 }
 
-func (j JobConfig) schedule(runner jobRunner) error {
+func (j JobConfig) shouldRun(t time.Time, lastCompleted *CompletedJob) (bool, error) {
 	if !j.Enabled {
-		return nil
-	}
-
-	lastCompleted, err := runner.lastCompleted(j.Name)
-	if err != nil {
-		return err
+		return false, nil
 	}
 
 	exitStatus := -1
@@ -52,31 +47,30 @@ func (j JobConfig) schedule(runner jobRunner) error {
 		started = int(lastCompleted.Started.Unix())
 	}
 
-	now := time.Now()
 	kvpairs := []starlark.Tuple{
 		starlark.Tuple{
 			starlark.String("minute"),
-			starlark.MakeInt(now.Minute()),
+			starlark.MakeInt(t.Minute()),
 		},
 		starlark.Tuple{
 			starlark.String("hour"),
-			starlark.MakeInt(now.Hour()),
+			starlark.MakeInt(t.Hour()),
 		},
 		starlark.Tuple{
 			starlark.String("day"),
-			starlark.MakeInt(now.Day()),
+			starlark.MakeInt(t.Day()),
 		},
 		starlark.Tuple{
 			starlark.String("month"),
-			starlark.MakeInt(int(now.Month())),
+			starlark.MakeInt(int(t.Month())),
 		},
 		starlark.Tuple{
 			starlark.String("dow"),
-			starlark.MakeInt(int(now.Weekday())),
+			starlark.MakeInt(int(t.Weekday())),
 		},
 		starlark.Tuple{
 			starlark.String("timestamp"),
-			starlark.MakeInt(int(now.Unix())),
+			starlark.MakeInt(int(t.Unix())),
 		},
 		starlark.Tuple{
 			starlark.String("exit_status"),
@@ -95,18 +89,35 @@ func (j JobConfig) schedule(runner jobRunner) error {
 	thread := &starlark.Thread{Name: "schedule"}
 	result, err := starlark.Call(thread, j.ShouldRun, nil, kvpairs)
 	if err != nil {
-		return fmt.Errorf(`failed to call "should_run": %v`, err)
+		return false, fmt.Errorf(`failed to call "should_run": %v`, err)
 	}
 
 	switch result {
 
 	case starlark.False:
+		return false, nil
 
 	case starlark.True:
-		runner.addJob(j)
+		return true, nil
 
 	default:
-		return fmt.Errorf(`"should_run" returned bad value: %v`, result)
+		return false, fmt.Errorf(`"should_run" returned bad value: %v`, result)
+	}
+}
+
+func (j JobConfig) addToQueueIfDue(runner jobRunner, t time.Time) error {
+	lastCompleted, err := runner.lastCompleted(j.Name)
+	if err != nil {
+		return err
+	}
+
+	shouldRun, err := j.shouldRun(t, lastCompleted)
+	if err != nil {
+		return err
+	}
+
+	if shouldRun {
+		runner.addJob(j)
 	}
 
 	return nil

jobscheduler.go

@@ -35,12 +35,12 @@ func newJobScheduler() jobScheduler {
 	}
 }
 
-func (jsc jobScheduler) scheduleOnce(runner jobRunner) error {
+func (jsc jobScheduler) addDueJobsToQueue(runner jobRunner, t time.Time) error {
 	jsc.mu.RLock()
 	defer jsc.mu.RUnlock()
 
 	for name, job := range jsc.byName {
-		err := job.schedule(runner)
+		err := job.addToQueueIfDue(runner, t)
 		if err != nil {
 			return newJobError(name, fmt.Errorf("scheduling error: %w", err))
 		}
@@ -53,15 +53,34 @@ func (jsc jobScheduler) schedule(runner jobRunner) error {
 	ticker := time.NewTicker(scheduleInterval)
 	defer ticker.Stop()
 
-	err := jsc.scheduleOnce(runner)
+	current := time.Now()
+	var last time.Time
+
+	err := jsc.addDueJobsToQueue(runner, current)
 	if err != nil {
 		return err
 	}
 
 	for range ticker.C {
-		err := jsc.scheduleOnce(runner)
-		if err != nil {
-			return err
+		last = current
+		current = time.Now()
+
+		// Account for missed time.
+		// Do not run missed jobs if more than maxMissedTime has elapsed.
+		// On an overloaded system, the ticker can miss a minute.
+		// For example, this may happen because Regular was swapped out.
+		// It would prevent Regular from running jobs scheduled for that minute and that minute alone.
+		// The purpose of this approach is to catch up on missed jobs.
+		// However, we should run days' worth of missed jobs after system hibernation.
+		if current.Sub(last) > maxMissedTime {
+			last = current
+		}
+
+		for t := last; t.Before(current); t = t.Add(time.Minute) {
+			err := jsc.addDueJobsToQueue(runner, t)
+			if err != nil {
+				return err
+			}
 		}
 	}
 

runcmd.go

@@ -3,6 +3,7 @@ package main
 import (
 	"fmt"
 	"path/filepath"
+	"time"
 )
 
 func (r *RunCmd) Run(config Config) error {
@@ -18,6 +19,7 @@ func (r *RunCmd) Run(config Config) error {
 	}
 
 	jobs := newJobScheduler()
+	now := time.Now()
 
 	for _, jobName := range r.JobNames {
 		path := filepath.Join(config.ConfigRoot, jobName, jobFileName)
@@ -32,7 +34,7 @@ func (r *RunCmd) Run(config Config) error {
 		if r.Force {
 			runner.addJob(*job)
 		} else {
-			if err := job.schedule(runner); err != nil {
+			if err := job.addToQueueIfDue(runner, now); err != nil {
 				return fmt.Errorf("failed to schedule job %q: %w", job.Name, err)
 			}
 		}