Change project layout to make easier debug / tests"

This commit is contained in:
eleksir
2024-09-22 23:52:57 +03:00
parent 9d3bc186cc
commit 4bff21f4c7
112 changed files with 271 additions and 1683 deletions

View File

@@ -4,6 +4,7 @@ import "fmt"
// Public error definitions
var (
ErrCronJobInvalid = fmt.Errorf("gocron: CronJob: invalid crontab")
ErrCronJobParse = fmt.Errorf("gocron: CronJob: crontab parse failure")
ErrDailyJobAtTimeNil = fmt.Errorf("gocron: DailyJob: atTime within atTimes must not be nil")
ErrDailyJobAtTimesNil = fmt.Errorf("gocron: DailyJob: atTimes must not be nil")

View File

@@ -196,7 +196,7 @@ func (e *executor) start() {
default:
// runner is busy, reschedule the work for later
// which means we just skip it here and do nothing
// TODO when metrics are added, this should increment a rescheduled metric
e.incrementJobCounter(*j, SingletonRescheduled)
e.sendOutForRescheduling(&jIn)
}
} else {
@@ -397,9 +397,7 @@ func (e *executor) runJob(j internalJob, jIn jobIn) {
startTime := time.Now()
err := e.callJobWithRecover(j)
if e.monitor != nil {
e.monitor.RecordJobTiming(startTime, time.Now(), j.id, j.name, j.tags)
}
e.recordJobTiming(startTime, time.Now(), j)
if err != nil {
_ = callJobFuncWithParams(j.afterJobRunsWithError, j.id, j.name, err)
e.incrementJobCounter(j, Fail)
@@ -422,6 +420,12 @@ func (e *executor) callJobWithRecover(j internalJob) (err error) {
return callJobFuncWithParams(j.function, j.parameters...)
}
func (e *executor) recordJobTiming(start time.Time, end time.Time, j internalJob) {
if e.monitor != nil {
e.monitor.RecordJobTiming(start, end, j.id, j.name, j.tags)
}
}
func (e *executor) incrementJobCounter(j internalJob, status JobStatus) {
if e.monitor != nil {
e.monitor.IncrementJob(j.id, j.name, j.tags, status)

View File

@@ -116,7 +116,7 @@ type cronJobDefinition struct {
withSeconds bool
}
func (c cronJobDefinition) setup(j *internalJob, location *time.Location, _ time.Time) error {
func (c cronJobDefinition) setup(j *internalJob, location *time.Location, now time.Time) error {
var withLocation string
if strings.HasPrefix(c.crontab, "TZ=") || strings.HasPrefix(c.crontab, "CRON_TZ=") {
withLocation = c.crontab
@@ -140,6 +140,9 @@ func (c cronJobDefinition) setup(j *internalJob, location *time.Location, _ time
if err != nil {
return errors.Join(ErrCronJobParse, err)
}
if cronSchedule.Next(now).IsZero() {
return ErrCronJobInvalid
}
j.jobSchedule = &cronJob{cronSchedule: cronSchedule}
return nil
@@ -456,6 +459,8 @@ type oneTimeJobDefinition struct {
func (o oneTimeJobDefinition) setup(j *internalJob, _ *time.Location, now time.Time) error {
sortedTimes := o.startAt(j)
slices.SortStableFunc(sortedTimes, ascendingTime)
// deduplicate the times
sortedTimes = removeSliceDuplicatesTimeOnSortedSlice(sortedTimes)
// keep only schedules that are in the future
idx, found := slices.BinarySearchFunc(sortedTimes, now, ascendingTime)
if found {
@@ -469,6 +474,16 @@ func (o oneTimeJobDefinition) setup(j *internalJob, _ *time.Location, now time.T
return nil
}
func removeSliceDuplicatesTimeOnSortedSlice(times []time.Time) []time.Time {
ret := make([]time.Time, 0, len(times))
for i, t := range times {
if i == 0 || t != times[i-1] {
ret = append(ret, t)
}
}
return ret
}
// OneTimeJobStartAtOption defines when the one time job is run
type OneTimeJobStartAtOption func(*internalJob) []time.Time

View File

@@ -11,9 +11,10 @@ type JobStatus string
// The different statuses of job that can be used.
const (
Fail JobStatus = "fail"
Success JobStatus = "success"
Skip JobStatus = "skip"
Fail JobStatus = "fail"
Success JobStatus = "success"
Skip JobStatus = "skip"
SingletonRescheduled JobStatus = "singleton_rescheduled"
)
// Monitor represents the interface to collect jobs metrics.

View File

@@ -311,7 +311,7 @@ func (s *scheduler) selectRemoveJob(id uuid.UUID) {
}
// Jobs coming back from the executor to the scheduler that
// need to evaluated for rescheduling.
// need to be evaluated for rescheduling.
func (s *scheduler) selectExecJobsOutForRescheduling(id uuid.UUID) {
select {
case <-s.shutdownCtx.Done():
@@ -359,6 +359,13 @@ func (s *scheduler) selectExecJobsOutForRescheduling(id uuid.UUID) {
next = j.next(next)
}
}
// Clean up any existing timer to prevent leaks
if j.timer != nil {
j.timer.Stop()
j.timer = nil // Ensure timer is cleared for GC
}
j.nextScheduled = append(j.nextScheduled, next)
j.timer = s.exec.clock.AfterFunc(next.Sub(s.now()), func() {
// set the actual timer on the job here and listen for