Repository: furiko-io/furiko (go)

Issue:
Bug: CronController performs back-scheduling when re-enabling JobConfig
The CronController will perform back-scheduling based on the `status.lastScheduleTime` and `spec.schedule.constraints.notBefore`. We will always `notBefore` to the current time if we detect that the schedule is updated, to avoid wrongly back-scheduling when the user shortens the cron schedule (e.g. every 00:00 changed to start of every hour, do not schedule until start of next hour).

However, the case when we update `spec.schedule.disabled` from `true` to `false` is not handled, causing it to still back-schedule incorrectly.

Pull request description:
feat(execution): Bump LastUpdated on schedule update
Closes #50.

- Separate `spec.schedule.constraints.notBefore` to `spec.schedule.lastUpdated`, to avoid conflation of server-managed field vs user-defined fields.
- Add `notAfter` constraint, the dual of `notBefore`.
- Fixes JobConfigMutatingWebhook to handle case when `cron` was not updated but `disabled` was updated.

Patch:
diff --git a/apis/execution/v1alpha1/jobconfig_types.go b/apis/execution/v1alpha1/jobconfig_types.go
index fc8ad57..5ee26ea 100644
--- a/apis/execution/v1alpha1/jobconfig_types.go
+++ b/apis/execution/v1alpha1/jobconfig_types.go
@@ -65,6 +65,16 @@ type ScheduleSpec struct {
 	//
 	// +optional
 	Constraints *ScheduleContraints `json:"constraints,omitempty"`
+
+	// Specifies the time that the schedule was last upated. This prevents
+	// accidental back-scheduling.
+	//
+	// For example, if a JobConfig that was previously disabled from automatic
+	// scheduling is now enabled, we do not want to perform back-scheduling for
+	// schedules after LastScheduled prior to updating of the JobConfig.
+	//
+	// +optional
+	LastUpdated *metav1.Time `json:"lastUpdated,omitempty"`
 }
 
 // CronSchedule defines a Schedule based on cron format.
@@ -101,17 +111,17 @@ type CronSchedule struct {
 
 // ScheduleContraints defines constraints for automatic scheduling.
 type ScheduleContraints struct {
-	// If set, the scheduler should not perform any scheduling for timestamps before
-	// this.
-	//
-	// For example, if a JobConfig that was previously disabled from automatic
-	// scheduling is now enabled, then NotBefore will be automatically updated to
-	// the current time. This prevents accidental back-scheduling so that we may not
-	// use LastScheduleTime to compute back-schedules that are before it was
-	// re-enabled.
+	// Specifies the earliest possible time that is allowed to be scheduled. If set,
+	// the scheduler should not create schedules before this time.
 	//
 	// +optional
 	NotBefore *metav1.Time `json:"notBefore,omitempty"`
+
+	// Specifies the latest possible time that is allowed to be scheduled. If set,
+	// the scheduler should not create schedules after this time.
+	//
+	// +optional
+	NotAfter *metav1.Time `json:"notAfter,omitempty"`
 }
 
 type ConcurrencyPolicy string
@@ -405,23 +415,10 @@ type JobConfigStatus struct {
 	// during controller downtime. If the controller was down for a short period of
 	// time, any schedules that were missed during the downtime will be
 	// back-scheduled, subject to the number of schedules missed since
-	// LastScheduleTime.
-	//
-	// To prevent thundering herd effects, there are two controller configuration
-	// values that will restrict the number of back-scheduled jobs created after the
-	// controller comes back up:
-	//
-	//  1. MaxMissedSchedules: Defines a maximum number of jobs that the controller
-	//     should attempt after coming back up. Having a sane value here would prevent a
-	//     thundering herd of jobs being scheduled that would exhaust resources in the
-	//     cluster. Defaults to 5 if not defined.
-	//  2. MaxDowntimeThresholdSeconds: Defines the maximum downtime threshold that
-	//     the controller can tolerate. If the controller was intentionally shut down
-	//     for an extended period of time, we should not attempt to back-schedule jobs
-	//     once it was started. Defaults to 5 minutes if not defined.
+	// LastScheduled.
 	//
 	// +optional
-	LastScheduleTime *metav1.Time `json:"lastScheduleTime,omitempty"`
+	LastScheduled *metav1.Time `json:"lastScheduled,omitempty"`
 }
 
 type JobConfigState string
diff --git a/apis/execution/v1alpha1/zz_generated.deepcopy.go b/apis/execution/v1alpha1/zz_generated.deepcopy.go
index de69105..4129c3d 100644
--- a/apis/execution/v1alpha1/zz_generated.deepcopy.go
+++ b/apis/execution/v1alpha1/zz_generated.deepcopy.go
@@ -321,8 +321,8 @@ func (in *JobConfigStatus) DeepCopyInto(out *JobConfigStatus) {
 		*out = make([]JobReference, len(*in))
 		copy(*out, *in)
 	}
-	if in.LastScheduleTime != nil {
-		in, out := &in.LastScheduleTime, &out.LastScheduleTime
+	if in.LastScheduled != nil {
+		in, out := &in.LastScheduled, &out.LastScheduled
 		*out = (*in).DeepCopy()
 	}
 }
@@ -610,6 +610,10 @@ func (in *ScheduleContraints) DeepCopyInto(out *ScheduleContraints) {
 		in, out := &in.NotBefore, &out.NotBefore
 		*out = (*in).DeepCopy()
 	}
+	if in.NotAfter != nil {
+		in, out := &in.NotAfter, &out.NotAfter
+		*out = (*in).DeepCopy()
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ScheduleContraints.
@@ -635,6 +639,10 @@ func (in *ScheduleSpec) DeepCopyInto(out *ScheduleSpec) {
 		*out = new(ScheduleContraints)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.LastUpdated != nil {
+		in, out := &in.LastUpdated, &out.LastUpdated
+		*out = (*in).DeepCopy()
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ScheduleSpec.
diff --git a/config/crd/bases/execution.furiko.io_jobconfigs.yaml b/config/crd/bases/execution.furiko.io_jobconfigs.yaml
index eb8ef9f..f06222b 100644
--- a/config/crd/bases/execution.furiko.io_jobconfigs.yaml
+++ b/config/crd/bases/execution.furiko.io_jobconfigs.yaml
@@ -170,8 +170,12 @@ spec:
                     constraints:
                       description: Specifies any constraints that should apply to this Schedule.
                       properties:
+                        notAfter:
+                          description: Specifies the latest possible time that is allowed to be scheduled. If set, the scheduler should not create schedules after this time.
+                          format: date-time
+                          type: string
                         notBefore:
-                          description: "If set, the scheduler should not perform any scheduling for timestamps before this. \n For example, if a JobConfig that was previously disabled from automatic scheduling is now enabled, then NotBefore will be automatically updated to the current time. This prevents accidental back-scheduling so that we may not use LastScheduleTime to compute back-schedules that are before it was re-enabled."
+                          description: Specifies the earliest possible time that is allowed to be scheduled. If set, the scheduler should not create schedules before this time.
                           format: date-time
                           type: string
                       type: object
@@ -190,6 +194,10 @@ spec:
                     disabled:
                       description: If true, then automatic scheduling will be disabled for the JobConfig.
                       type: boolean
+                    lastUpdated:
+                      description: "Specifies the time that the schedule was last upated. This prevents accidental back-scheduling. \n For example, if a JobConfig that was previously disabled from automatic scheduling is now enabled, we do not want to perform back-scheduling for schedules after LastScheduled prior to updating of the JobConfig."
+                      format: date-time
+                      type: string
                   type: object
                 template:
                   description: Template for creating the Job.
@@ -4304,8 +4312,8 @@ spec:
                       - namespace
                     type: object
                   type: array
-                lastScheduleTime:
-                  description: "The last known schedule time for this job config, used to persist state during controller downtime. If the controller was down for a short period of time, any schedules that were missed during the downtime will be back-scheduled, subject to the number of schedules missed since LastScheduleTime. \n To prevent thundering herd effects, there are two controller configuration values that will restrict the number of back-scheduled jobs created after the controller comes back up: \n 1. MaxMissedSchedules: Defines a maximum number of jobs that the controller should attempt after coming back up. Having a sane value here would prevent a thundering herd of jobs being scheduled that would exhaust resources in the cluster. Defaults to 5 if not defined. 2. MaxDowntimeThresholdSeconds: Defines the maximum downtime threshold that the controller can tolerate. If the controller was intentionally shut down for an extended period of time, we should not attempt to back-schedule jobs once it was started. Defaults to 5 minutes if not defined."
+                lastScheduled:
+                  description: The last known schedule time for this job config, used to persist state during controller downtime. If the controller was down for a short period of time, any schedules that were missed during the downtime will be back-scheduled, subject to the number of schedules missed since LastScheduled.
                   format: date-time
                   type: string
                 queued:
diff --git a/pkg/execution/controllers/croncontroller/schedule.go b/pkg/execution/controllers/croncontroller/schedule.go
index 9c7dda7..6fee950 100644
--- a/pkg/execution/controllers/croncontroller/schedule.go
+++ b/pkg/execution/controllers/croncontroller/schedule.go
@@ -77,7 +77,7 @@ func (w *Schedule) GetNextScheduleTime(
 	// Did not previously compute next schedule time (or was cleared). Use the last
 	// schedule time in the JobConfig's status to back-schedule all missed schedules
 	// (up to a threshold).
-	lastScheduleTime := jobConfig.Status.LastScheduleTime
+	lastScheduleTime := jobConfig.Status.LastScheduled
 	if lastScheduleTime != nil && !lastScheduleTime.IsZero() {
 		if Clock.Since(lastScheduleTime.Time) > maxDowntimeThreshold {
 			newLastScheduleTime := metav1.NewTime(Clock.Now().Add(-maxDowntimeThreshold))
@@ -86,30 +86,43 @@ func (w *Schedule) GetNextScheduleTime(
 		fromTime = lastScheduleTime.Time
 	}
 
-	// Cannot schedule before schedule.NotBefore.
+	// Bump to LastUpdated to avoid accidental back-scheduling.
+	if spec := jobConfig.Spec.Schedule; spec != nil {
+		if lastUpdated := spec.LastUpdated; !lastUpdated.IsZero() && lastUpdated.After(fromTime) {
+			fromTime = lastUpdated.Time
+		}
+	}
+
+	// Cannot schedule before NotBefore.
+	// NOTE(irvinlim): Compute next using fromTime, so we sub nanosecond in case time falls exactly on NotBefore
 	if spec := jobConfig.Spec.Schedule; spec != nil && spec.Constraints != nil {
-		if nbf := spec.Constraints.NotBefore; !nbf.IsZero() && nbf.After(fromTime) {
-			fromTime = nbf.Time
+		if nbf := spec.Constraints.NotBefore; !nbf.IsZero() && fromTime.Before(nbf.Time) {
+			fromTime = nbf.Time.Add(-time.Nanosecond)
 		}
 	}
 
 	// Set timezone correctly.
 	fromTime = fromTime.In(timezone)
 
-	// Compute the next scheduled time.
-	next := expr.Next(fromTime)
-
-	// We update our internal map so it will be fetched when the time comes.
-	w.setNextScheduleTime(jobConfig, next)
-	return next
+	// Bump and save the next schedule time.
+	return w.BumpNextScheduleTime(jobConfig, fromTime, expr)
 }
 
 // BumpNextScheduleTime updates the next schedule time of a JobConfig.
 func (w *Schedule) BumpNextScheduleTime(
 	jobConfig *execution.JobConfig, fromTime time.Time, expr *cronexpr.Expression,
-) {
+) time.Time {
 	next := expr.Next(fromTime)
+
+	// Cannot schedule after NotAfter.
+	if spec := jobConfig.Spec.Schedule; spec != nil && spec.Constraints != nil {
+		if naf := spec.Constraints.NotAfter; !naf.IsZero() && next.After(naf.Time) {
+			next = time.Time{}
+		}
+	}
+
 	w.setNextScheduleTime(jobConfig, next)
+	return next
 }
 
 // FlushNextScheduleTime flushes the next schedule time for a JobConfig, causing it to be recomputed on
diff --git a/pkg/execution/controllers/jobconfigcontroller/reconciler.go b/pkg/execution/controllers/jobconfigcontroller/reconciler.go
index dcbc691..d4a3ed6 100644
--- a/pkg/execution/controllers/jobconfigcontroller/reconciler.go
+++ b/pkg/execution/controllers/jobconfigcontroller/reconciler.go
@@ -111,7 +111,7 @@ func (w *Reconciler) SyncOne(ctx context.Context, namespace, name string, _ int)
 
 	// Update last schedule time.
 	if lastScheduleTime := jobconfig.GetLastScheduleTime(rjs); !lastScheduleTime.IsZero() {
-		newRjc.Status.LastScheduleTime = ktime.TimeMax(lastScheduleTime, newRjc.Status.LastScheduleTime)
+		newRjc.Status.LastScheduled = ktime.TimeMax(lastScheduleTime, newRjc.Status.LastScheduled)
 	}
 
 	// Compute final state.
diff --git a/pkg/execution/mutation/mutation.go b/pkg/execution/mutation/mutation.go
index 35c211c..4bc26b5 100644
--- a/pkg/execution/mutation/mutation.go
+++ b/pkg/execution/mutation/mutation.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -37,7 +38,6 @@ import (
 	"github.com/furiko-io/furiko/pkg/execution/variablecontext"
 	executionlister "github.com/furiko-io/furiko/pkg/generated/listers/execution/v1alpha1"
 	"github.com/furiko-io/furiko/pkg/runtime/controllercontext"
-	"github.com/furiko-io/furiko/pkg/utils/cmp"
 	"github.com/furiko-io/furiko/pkg/utils/execution/jobconfig"
 	"github.com/furiko-io/furiko/pkg/utils/jsonyaml"
 	"github.com/furiko-io/furiko/pkg/utils/k8sutils"
@@ -78,15 +78,12 @@ func (m *Mutator) MutateJobConfig(rjc *v1alpha1.JobConfig) *webhook.Result {
 func (m *Mutator) MutateCreateJobConfig(rjc *v1alpha1.JobConfig) *webhook.Result {
 	result := webhook.NewResult()
 
-	// Bump NotBefore when JobConfig is first created.
+	// Bump LastUpdated when JobConfig is first created.
 	// NOTE(irvinlim): Do not create scheduleSpec if nil.
 	now := metav1.NewTime(Clock.Now())
 	if rjc.Spec.Schedule != nil {
-		if rjc.Spec.Schedule.Constraints == nil {
-			rjc.Spec.Schedule.Constraints = &v1alpha1.ScheduleContraints{}
-		}
-		if !ktime.IsTimeSetAndLaterThan(rjc.Spec.Schedule.Constraints.NotBefore, now.Time) {
-			rjc.Spec.Schedule.Constraints.NotBefore = &now
+		if !ktime.IsTimeSetAndLaterThan(rjc.Spec.Schedule.LastUpdated, now.Time) {
+			rjc.Spec.Schedule.LastUpdated = &now
 		}
 	}
 
@@ -98,22 +95,15 @@ func (m *Mutator) MutateUpdateJobConfig(oldRjc, rjc *v1alpha1.JobConfig) *webhoo
 	result := webhook.NewResult()
 	now := metav1.NewTime(Clock.Now())
 
-	// Bump NotBefore if the cron schedules were updated.
+	// Bump LastUpdated if the schedule was updated.
 	if rjc.Spec.Schedule != nil {
-		var oldCron *v1alpha1.CronSchedule
-		if spec := oldRjc.Spec.Schedule; spec != nil {
-			oldCron = spec.Cron
+		newSchedule := rjc.Spec.Schedule.DeepCopy()
+		if oldRjc.Spec.Schedule != nil {
+			newSchedule.LastUpdated = oldRjc.Spec.Schedule.LastUpdated
 		}
-		isEqual, err := cmp.IsJSONEqual(oldCron, rjc.Spec.Schedule.Cron)
-		if err != nil {
-			fldPath := field.NewPath("spec").Child("schedule").Child("cron")
-			result.Errors = append(result.Errors, field.InternalError(fldPath, err))
-		} else if !isEqual {
-			if rjc.Spec.Schedule.Constraints == nil {
-				rjc.Spec.Schedule.Constraints = &v1alpha1.ScheduleContraints{}
-			}
-			if !ktime.IsTimeSetAndLaterThan(rjc.Spec.Schedule.Constraints.NotBefore, now.Time) {
-				rjc.Spec.Schedule.Constraints.NotBefore = &now
+		if !apiequality.Semantic.DeepEqual(oldRjc.Spec.Schedule, newSchedule) {
+			if !ktime.IsTimeSetAndLaterThan(rjc.Spec.Schedule.LastUpdated, now.Time) {
+				rjc.Spec.Schedule.LastUpdated = &now
 			}
 		}
 	}


Test patch:
diff --git a/pkg/execution/controllers/croncontroller/schedule_test.go b/pkg/execution/controllers/croncontroller/schedule_test.go
index b93fc5c..4a1f3bc 100644
--- a/pkg/execution/controllers/croncontroller/schedule_test.go
+++ b/pkg/execution/controllers/croncontroller/schedule_test.go
@@ -38,8 +38,11 @@ const (
 )
 
 var (
-	lastScheduleTime     = metav1.NewTime(testutils.Mktime("2021-02-09T04:02:00Z"))
-	longLastScheduleTime = metav1.NewTime(testutils.Mktime("2021-02-09T02:46:00Z"))
+	lastScheduleTime     = testutils.Mkmtime("2021-02-09T04:02:00Z")
+	longLastScheduleTime = testutils.Mkmtime("2021-02-09T02:46:00Z")
+	notBeforeTime        = testutils.Mkmtime("2021-02-09T04:01:00Z")
+	notAfterTime         = testutils.Mkmtime("2021-02-09T04:08:00Z")
+	futureTime           = testutils.Mkmtime("2021-02-10T04:02:00Z")
 
 	jobConfigEveryMinute = &execution.JobConfig{
 		ObjectMeta: metav1.ObjectMeta{
@@ -79,7 +82,7 @@ var (
 			},
 		},
 		Status: execution.JobConfigStatus{
-			LastScheduleTime: testutils.Mkmtimep("2021-10-27T22:15:00Z"),
+			LastScheduled: testutils.Mkmtimep("2021-10-27T22:15:00Z"),
 		},
 	}
 
@@ -95,7 +98,7 @@ var (
 			},
 		},
 		Status: execution.JobConfigStatus{
-			LastScheduleTime: &lastScheduleTime,
+			LastScheduled: &lastScheduleTime,
 		},
 	}
 
@@ -111,7 +114,7 @@ var (
 			},
 		},
 		Status: execution.JobConfigStatus{
-			LastScheduleTime: &longLastScheduleTime,
+			LastScheduled: &longLastScheduleTime,
 		},
 	}
 
@@ -142,13 +145,121 @@ var (
 			},
 		},
 		Status: execution.JobConfigStatus{
-			LastScheduleTime: testutils.Mkmtimep("2021-06-24T09:00:00Z"),
+			LastScheduled: testutils.Mkmtimep("2021-06-24T09:00:00Z"),
 		},
 	}
 
-	jobConfigWithLastScheduleTimeAndNotBefore = &execution.JobConfig{
+	jobConfigWithLastScheduledAndUpdated = &execution.JobConfig{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: "job-config-with-last-schedule-time-and-not-before",
+			Name: "job-config-with-last-schedule-and-updated",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "* * * * *",
+				},
+				LastUpdated: testutils.Mkmtimep(now),
+			},
+		},
+		Status: execution.JobConfigStatus{
+			LastScheduled: &longLastScheduleTime,
+		},
+	}
+
+	jobConfigWithNotBeforeConstraint = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-with-not-before-constraint",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "* * * * *",
+				},
+				Constraints: &execution.ScheduleContraints{
+					NotBefore: &futureTime,
+				},
+			},
+		},
+	}
+
+	jobConfigWithNotBeforeAndLastScheduled = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-with-not-before-and-last-scheduled",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "* * * * *",
+				},
+				Constraints: &execution.ScheduleContraints{
+					NotBefore: &futureTime,
+				},
+			},
+		},
+		Status: execution.JobConfigStatus{
+			LastScheduled: &lastScheduleTime,
+		},
+	}
+
+	jobConfigWithNotBeforeAndLastUpdated = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-with-not-before-and-last-updated",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "* * * * *",
+				},
+				Constraints: &execution.ScheduleContraints{
+					NotBefore: &notBeforeTime,
+				},
+				LastUpdated: testutils.Mkmtimep(now),
+			},
+		},
+		Status: execution.JobConfigStatus{
+			LastScheduled: &lastScheduleTime,
+		},
+	}
+
+	jobConfigWithFutureNotBeforeAndLastUpdated = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-with-future-not-before-and-last-updated",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "* * * * *",
+				},
+				Constraints: &execution.ScheduleContraints{
+					NotBefore: &futureTime,
+				},
+				LastUpdated: testutils.Mkmtimep(now),
+			},
+		},
+		Status: execution.JobConfigStatus{
+			LastScheduled: &lastScheduleTime,
+		},
+	}
+
+	jobConfigWithNotAfterConstraint = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-with-not-after-constraint",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "* * * * *",
+				},
+				Constraints: &execution.ScheduleContraints{
+					NotAfter: &notAfterTime,
+				},
+			},
+		},
+	}
+
+	jobConfigWithNotAfterWithLastScheduled = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-with-not-after-and-last-scheduled",
 		},
 		Spec: execution.JobConfigSpec{
 			Schedule: &execution.ScheduleSpec{
@@ -156,12 +267,32 @@ var (
 					Expression: "* * * * *",
 				},
 				Constraints: &execution.ScheduleContraints{
-					NotBefore: testutils.Mkmtimep(now),
+					NotAfter: &notAfterTime,
 				},
 			},
 		},
 		Status: execution.JobConfigStatus{
-			LastScheduleTime: &longLastScheduleTime,
+			LastScheduled: &lastScheduleTime,
+		},
+	}
+
+	jobConfigWithNotAfterWithLastUpdated = &execution.JobConfig{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "job-config-with-not-after-and-last-updated",
+		},
+		Spec: execution.JobConfigSpec{
+			Schedule: &execution.ScheduleSpec{
+				Cron: &execution.CronSchedule{
+					Expression: "* * * * *",
+				},
+				Constraints: &execution.ScheduleContraints{
+					NotAfter: &notAfterTime,
+				},
+				LastUpdated: testutils.Mkmtimep(now),
+			},
+		},
+		Status: execution.JobConfigStatus{
+			LastScheduled: &lastScheduleTime,
 		},
 	}
 )
@@ -191,7 +322,7 @@ func TestSchedule(t *testing.T) {
 			},
 		},
 		{
-			name:      "No future schedules with LastScheduleTime",
+			name:      "No future schedules with LastScheduled",
 			jobConfig: jobConfigPointInTimeWithAlreadySetLastScheduleTime,
 			fromTime:  testutils.Mktime("2021-10-28T06:00:00+08:00"),
 			cases: []time.Time{
@@ -199,7 +330,7 @@ func TestSchedule(t *testing.T) {
 			},
 		},
 		{
-			name:      "Start from LastScheduleTime",
+			name:      "Start from LastScheduled",
 			jobConfig: jobConfigWithLastScheduleTime,
 			cases: []time.Time{
 				testutils.Mktime("2021-02-09T04:03:00Z"),
@@ -209,7 +340,7 @@ func TestSchedule(t *testing.T) {
 			},
 		},
 		{
-			name:      "Enforce maximum LastScheduleTime ago",
+			name:      "Enforce maximum LastScheduled ago",
 			jobConfig: jobConfigWithLastScheduleTimeLongAgo,
 			cases: []time.Time{
 				testutils.Mktime("2021-02-09T04:02:00Z"),
@@ -238,14 +369,81 @@ func TestSchedule(t *testing.T) {
 			},
 		},
 		{
-			name:      "Respect ScheduleNotBeforeTimestamp",
-			jobConfig: jobConfigWithLastScheduleTimeAndNotBefore,
+			name:      "Do not back-schedule with LastUpdated",
+			jobConfig: jobConfigWithLastScheduledAndUpdated,
+			cases: []time.Time{
+				testutils.Mktime("2021-02-09T04:07:00Z"),
+				testutils.Mktime("2021-02-09T04:08:00Z"),
+				testutils.Mktime("2021-02-09T04:09:00Z"),
+			},
+		},
+		{
+			name:      "Respect NotBefore constraint",
+			jobConfig: jobConfigWithNotBeforeConstraint,
+			cases: []time.Time{
+				testutils.Mktime("2021-02-10T04:02:00Z"),
+				testutils.Mktime("2021-02-10T04:03:00Z"),
+				testutils.Mktime("2021-02-10T04:04:00Z"),
+			},
+		},
+		{
+			name:      "Respect NotBefore constraint with LastScheduled",
+			jobConfig: jobConfigWithNotBeforeAndLastScheduled,
+			cases: []time.Time{
+				testutils.Mktime("2021-02-10T04:02:00Z"),
+				testutils.Mktime("2021-02-10T04:03:00Z"),
+				testutils.Mktime("2021-02-10T04:04:00Z"),
+			},
+		},
+		{
+			name:      "Respect NotBefore constraint with LastUpdated",
+			jobConfig: jobConfigWithNotBeforeAndLastUpdated,
 			cases: []time.Time{
 				testutils.Mktime("2021-02-09T04:07:00Z"),
 				testutils.Mktime("2021-02-09T04:08:00Z"),
 				testutils.Mktime("2021-02-09T04:09:00Z"),
 			},
 		},
+		{
+			name:      "Respect future NotBefore constraint with LastUpdated",
+			jobConfig: jobConfigWithFutureNotBeforeAndLastUpdated,
+			cases: []time.Time{
+				testutils.Mktime("2021-02-10T04:02:00Z"),
+				testutils.Mktime("2021-02-10T04:03:00Z"),
+				testutils.Mktime("2021-02-10T04:04:00Z"),
+			},
+		},
+		{
+			name:      "Respect NotAfter constraint",
+			jobConfig: jobConfigWithNotAfterConstraint,
+			cases: []time.Time{
+				testutils.Mktime("2021-02-09T04:07:00Z"),
+				testutils.Mktime("2021-02-09T04:08:00Z"),
+				{}, // no more
+			},
+		},
+		{
+			name:      "Respect NotAfter constraint with LastScheduled",
+			jobConfig: jobConfigWithNotAfterWithLastScheduled,
+			cases: []time.Time{
+				testutils.Mktime("2021-02-09T04:03:00Z"),
+				testutils.Mktime("2021-02-09T04:04:00Z"),
+				testutils.Mktime("2021-02-09T04:05:00Z"),
+				testutils.Mktime("2021-02-09T04:06:00Z"),
+				testutils.Mktime("2021-02-09T04:07:00Z"),
+				testutils.Mktime("2021-02-09T04:08:00Z"),
+				{}, // no more
+			},
+		},
+		{
+			name:      "Respect NotAfter constraint with LastUpdated",
+			jobConfig: jobConfigWithNotAfterWithLastUpdated,
+			cases: []time.Time{
+				testutils.Mktime("2021-02-09T04:07:00Z"),
+				testutils.Mktime("2021-02-09T04:08:00Z"),
+				{}, // no more
+			},
+		},
 	}
 	for _, tt := range tests {
 		tt := tt
diff --git a/pkg/execution/mutation/mutation_test.go b/pkg/execution/mutation/mutation_test.go
index 40a9071..ef63a25 100644
--- a/pkg/execution/mutation/mutation_test.go
+++ b/pkg/execution/mutation/mutation_test.go
@@ -328,7 +328,7 @@ func TestMutator_MutateCreateJobConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "bump NotBefore",
+			name: "bump LastUpdatedTime with updated cron schedule",
 			setup: func() {
 				mutation.Clock = clock.NewFakeClock(mockNow.Time)
 			},
@@ -348,7 +348,7 @@ func TestMutator_MutateCreateJobConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "don't reset future NotBefore",
+			name: "don't reset future LastUpdated",
 			setup: func() {
 				mutation.Clock = clock.NewFakeClock(mockNow.Time)
 			},
@@ -402,7 +402,7 @@ func TestMutator_MutateUpdateJobConfig(t *testing.T) {
 		setup        func()
 	}{
 		{
-			name: "don't add NotBefore without schedule",
+			name: "don't add LastUpdated without schedule",
 			oldRjc: &v1alpha1.JobConfig{
 				ObjectMeta: objectMetaJobConfig,
 				Spec: v1alpha1.JobConfigSpec{
@@ -417,7 +417,7 @@ func TestMutator_MutateUpdateJobConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "don't add NotBefore when removing schedule",
+			name: "don't add LastUpdated when removing schedule",
 			oldRjc: &v1alpha1.JobConfig{
 				ObjectMeta: objectMetaJobConfig,
 				Spec: v1alpha1.JobConfigSpec{
@@ -471,7 +471,7 @@ func TestMutator_MutateUpdateJobConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "bump NotBefore",
+			name: "bump LastUpdated",
 			setup: func() {
 				mutation.Clock = clock.NewFakeClock(mockNow.Time)
 			},
@@ -498,7 +498,7 @@ func TestMutator_MutateUpdateJobConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "bump NotBefore add schedule",
+			name: "bump LastUpdated add schedule",
 			setup: func() {
 				mutation.Clock = clock.NewFakeClock(mockNow.Time)
 			},
@@ -524,7 +524,45 @@ func TestMutator_MutateUpdateJobConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "don't reset future NotBefore",
+			name: "bump LastUpdated update disabled",
+			setup: func() {
+				mutation.Clock = clock.NewFakeClock(mockNow.Time)
+			},
+			oldRjc: &v1alpha1.JobConfig{
+				ObjectMeta: objectMetaJobConfig,
+				Spec: v1alpha1.JobConfigSpec{
+					Template: jobTemplateSpecBasic,
+					Schedule: &v1alpha1.ScheduleSpec{
+						Cron: &v1alpha1.CronSchedule{
+							Expression: cronSchedule1,
+							Timezone:   "Asia/Singapore",
+						},
+						Disabled: true,
+					},
+				},
+			},
+			rjc: &v1alpha1.JobConfig{
+				ObjectMeta: objectMetaJobConfig,
+				Spec: v1alpha1.JobConfigSpec{
+					Template: jobTemplateSpecBasic,
+					Schedule: &v1alpha1.ScheduleSpec{
+						Cron: &v1alpha1.CronSchedule{
+							Expression: cronSchedule1,
+							Timezone:   "Asia/Singapore",
+						},
+					},
+				},
+			},
+			want: &v1alpha1.JobConfig{
+				ObjectMeta: objectMetaJobConfig,
+				Spec: v1alpha1.JobConfigSpec{
+					Template: jobTemplateSpecBasic,
+					Schedule: mkScheduleSpec(cronSchedule1, &mockNow),
+				},
+			},
+		},
+		{
+			name: "don't reset future LastUpdated",
 			setup: func() {
 				mutation.Clock = clock.NewFakeClock(mockNow.Time)
 			},
@@ -1281,15 +1319,13 @@ func optionValues(values map[string]interface{}) string {
 	return string(bytes)
 }
 
-func mkScheduleSpec(expr string, nbf *metav1.Time) *v1alpha1.ScheduleSpec {
+func mkScheduleSpec(expr string, lastUpdated *metav1.Time) *v1alpha1.ScheduleSpec {
 	scheduleSpec := &v1alpha1.ScheduleSpec{
 		Cron: &v1alpha1.CronSchedule{
 			Expression: expr,
 			Timezone:   "Asia/Singapore",
 		},
-		Constraints: &v1alpha1.ScheduleContraints{
-			NotBefore: nbf,
-		},
+		LastUpdated: lastUpdated,
 	}
 	return scheduleSpec
 }