generated from TBD54566975/tbd-project-template
-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add tests, fixed case where slice was mutable by multiple owners
- Loading branch information
Showing
6 changed files
with
276 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,225 @@ | ||
| package cronjobs | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "strconv" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "connectrpc.com/connect" | ||
| "github.com/TBD54566975/ftl/backend/controller/dal" | ||
| "github.com/TBD54566975/ftl/backend/controller/scheduledtask" | ||
| ftlv1 "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1" | ||
| "github.com/TBD54566975/ftl/backend/schema" | ||
| "github.com/TBD54566975/ftl/internal/cron" | ||
| "github.com/TBD54566975/ftl/internal/log" | ||
| "github.com/TBD54566975/ftl/internal/model" | ||
| "github.com/TBD54566975/ftl/internal/slices" | ||
| "github.com/alecthomas/assert/v2" | ||
| "github.com/benbjohnson/clock" | ||
| "github.com/jpillora/backoff" | ||
| ) | ||
|
|
||
| type mockDAL struct { | ||
| lock sync.Mutex | ||
| clock *clock.Mock | ||
| jobs []dal.CronJob | ||
| attemptCountMap map[string]int | ||
| } | ||
|
|
||
| func (d *mockDAL) GetCronJobs(ctx context.Context) ([]dal.CronJob, error) { | ||
| d.lock.Lock() | ||
| defer d.lock.Unlock() | ||
|
|
||
| return d.jobs, nil | ||
| } | ||
|
|
||
| func (d *mockDAL) CreateCronJob(ctx context.Context, deploymentKey model.DeploymentKey, module string, verb string, schedule string, startTime time.Time, nextExecution time.Time) (dal.CronJob, error) { | ||
| d.lock.Lock() | ||
| defer d.lock.Unlock() | ||
|
|
||
| job := dal.CronJob{ | ||
| DeploymentKey: deploymentKey, | ||
| Module: module, | ||
| Verb: verb, | ||
| Schedule: schedule, | ||
| StartTime: startTime, | ||
| NextExecution: nextExecution, | ||
| State: dal.JobStateIdle, | ||
| } | ||
| d.jobs = append(d.jobs, job) | ||
| return job, nil | ||
| } | ||
|
|
||
| func (d *mockDAL) indexForJob(job dal.CronJob) (int, error) { | ||
| for i, j := range d.jobs { | ||
| if j.DeploymentKey == job.DeploymentKey && j.Verb == job.Verb { | ||
| return i, nil | ||
| } | ||
| } | ||
| return -1, fmt.Errorf("job not found") | ||
| } | ||
|
|
||
| func (d *mockDAL) StartCronJobs(ctx context.Context, jobs []dal.CronJob) (jobMap map[dal.CronJob]bool, keysWithNoReplicas map[model.DeploymentKey]bool, err error) { | ||
| d.lock.Lock() | ||
| defer d.lock.Unlock() | ||
|
|
||
| jobMap = map[dal.CronJob]bool{} | ||
| now := (*d.clock).Now() | ||
|
|
||
| for _, inputJob := range jobs { | ||
| i, err := d.indexForJob(inputJob) | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } | ||
| job := d.jobs[i] | ||
| if !job.NextExecution.After(now) && job.State == dal.JobStateIdle { | ||
| job.State = dal.JobStateExecuting | ||
| job.StartTime = (*d.clock).Now() | ||
| d.jobs[i] = job | ||
| jobMap[job] = true | ||
| } else { | ||
| jobMap[job] = false | ||
| } | ||
| d.attemptCountMap[job.Verb]++ | ||
| } | ||
| return jobMap, map[model.DeploymentKey]bool{}, nil | ||
| } | ||
|
|
||
| func (d *mockDAL) EndCronJob(ctx context.Context, job dal.CronJob, next time.Time) (dal.CronJob, error) { | ||
| d.lock.Lock() | ||
| defer d.lock.Unlock() | ||
|
|
||
| i, err := d.indexForJob(job) | ||
| if err != nil { | ||
| return dal.CronJob{}, err | ||
| } | ||
| internalJob := d.jobs[i] | ||
| if internalJob.State != dal.JobStateExecuting { | ||
| return dal.CronJob{}, fmt.Errorf("job can not be stopped, it isnt running") | ||
| } | ||
| if internalJob.StartTime != job.StartTime { | ||
| return dal.CronJob{}, fmt.Errorf("job can not be stopped, start time does not match") | ||
| } | ||
| internalJob.State = dal.JobStateIdle | ||
| internalJob.NextExecution = next | ||
| d.jobs[i] = internalJob | ||
| return internalJob, nil | ||
| } | ||
|
|
||
| func (d *mockDAL) GetStaleCronJobs(ctx context.Context, duration time.Duration) ([]dal.CronJob, error) { | ||
| d.lock.Lock() | ||
| defer d.lock.Unlock() | ||
|
|
||
| return slices.Filter(d.jobs, func(job dal.CronJob) bool { | ||
| return (*d.clock).Now().After(job.StartTime.Add(duration)) | ||
| }), nil | ||
| } | ||
|
|
||
| type mockScheduler struct { | ||
| } | ||
|
|
||
| func (s *mockScheduler) Singleton(retry backoff.Backoff, job scheduledtask.Job) { | ||
| // do nothing | ||
| } | ||
|
|
||
| func (s *mockScheduler) Parallel(retry backoff.Backoff, job scheduledtask.Job) { | ||
| // do nothing | ||
| } | ||
|
|
||
| type mockExecutor struct { | ||
| verbCallCount map[string]int | ||
| lock sync.Mutex | ||
| clock *clock.Mock | ||
| } | ||
|
|
||
| func (e *mockExecutor) Call(ctx context.Context, req *connect.Request[ftlv1.CallRequest]) (*connect.Response[ftlv1.CallResponse], error) { | ||
| verbRef := schema.RefFromProto(req.Msg.Verb) | ||
|
|
||
| e.lock.Lock() | ||
| e.verbCallCount[verbRef.Name]++ | ||
| e.lock.Unlock() | ||
|
|
||
| return &connect.Response[ftlv1.CallResponse]{}, nil | ||
| } | ||
|
|
||
| type controller struct { | ||
| key model.ControllerKey | ||
| DAL DAL | ||
| cronJobs *Service | ||
| } | ||
|
|
||
| func TestService(t *testing.T) { | ||
| t.Parallel() | ||
| ctx := log.ContextWithNewDefaultLogger(context.Background()) | ||
| ctx, cancel := context.WithCancel(ctx) | ||
| t.Cleanup(cancel) | ||
|
|
||
| // var singletonCount atomic.Int64 | ||
| // var multiCount atomic.Int64 | ||
|
|
||
| config := Config{Timeout: time.Minute * 5} | ||
| clock := clock.NewMock() | ||
| mockDal := &mockDAL{ | ||
| clock: clock, | ||
| lock: sync.Mutex{}, | ||
| attemptCountMap: map[string]int{}, | ||
| } | ||
| scheduler := &mockScheduler{} | ||
| executor := &mockExecutor{ | ||
| verbCallCount: map[string]int{}, | ||
| lock: sync.Mutex{}, | ||
| clock: clock, | ||
| } | ||
|
|
||
| // initial jobs | ||
| for i := range 20 { | ||
| deploymentKey := model.NewDeploymentKey("initial") | ||
| now := clock.Now() | ||
| cronStr := "*/10 * * * * * *" | ||
| pattern, err := cron.Parse(cronStr) | ||
| assert.NoError(t, err) | ||
| next, err := cron.NextAfter(pattern, now, false) | ||
| assert.NoError(t, err) | ||
| _, err = mockDal.CreateCronJob(ctx, deploymentKey, "initial", fmt.Sprintf("verb%d", i), cronStr, now, next) | ||
| assert.NoError(t, err) | ||
| } | ||
|
|
||
| controllers := []*controller{} | ||
| for i := range 5 { | ||
| key := model.NewControllerKey("localhost", strconv.Itoa(8080+i)) | ||
| controller := &controller{ | ||
| key: key, | ||
| DAL: mockDal, | ||
| cronJobs: NewForTesting(ctx, key, config, mockDal, scheduler, executor, clock), | ||
| } | ||
| controllers = append(controllers, controller) | ||
| } | ||
|
|
||
| time.Sleep(time.Millisecond * 100) | ||
|
|
||
| for _, c := range controllers { | ||
| go func() { | ||
| c.cronJobs.UpdatedControllerList(ctx, slices.Map(controllers, func(ctrl *controller) dal.Controller { | ||
| return dal.Controller{ | ||
| Key: ctrl.key, | ||
| } | ||
| })) | ||
| _, _ = c.cronJobs.resetJobs(ctx) | ||
| }() | ||
| } | ||
|
|
||
| clock.Add(time.Second * 5) | ||
| time.Sleep(time.Millisecond * 100) | ||
| for range 3 { | ||
| clock.Add(time.Second * 10) | ||
| time.Sleep(time.Millisecond * 100) | ||
| } | ||
|
|
||
| for _, j := range mockDal.jobs { | ||
| count := executor.verbCallCount[j.Verb] | ||
| assert.Equal(t, count, 3, "expected verb %s to be called 3 times", j.Verb) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.