mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Fix data race in scheduler (#9507)
## What changed? WISOTT ## Why? Data race is causing CI to break ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks NA
This commit is contained in:
@@ -103,10 +103,9 @@ func (s *workerComponent) DedicatedWorkerOptions(ns *namespace.Namespace) *worke
|
||||
}
|
||||
|
||||
func (s *workerComponent) Register(registry sdkworker.Registry, ns *namespace.Namespace, details workercommon.RegistrationDetails) func() {
|
||||
CurrentTweakablePolicies.EnableCHASMMigration = s.enableCHASMMigration(ns.Name().String())
|
||||
|
||||
enableMigration := s.enableCHASMMigration(ns.Name().String())
|
||||
wfFunc := func(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
|
||||
return schedulerWorkflowWithSpecBuilder(ctx, args, s.specBuilder)
|
||||
return schedulerWorkflowWithSpecBuilder(ctx, args, s.specBuilder, enableMigration)
|
||||
}
|
||||
registry.RegisterWorkflowWithOptions(wfFunc, workflow.RegisterOptions{Name: WorkflowType})
|
||||
|
||||
|
||||
@@ -112,8 +112,9 @@ type (
|
||||
// SpecBuilder is technically a non-deterministic dependency, but it's safe as
|
||||
// long as we only call methods on cspec inside of SideEffect (or in a query
|
||||
// without modifying state).
|
||||
specBuilder *SpecBuilder
|
||||
cspec *CompiledSpec
|
||||
specBuilder *SpecBuilder
|
||||
cspec *CompiledSpec
|
||||
enableCHASMMigration bool
|
||||
|
||||
tweakables TweakablePolicies
|
||||
|
||||
@@ -221,10 +222,10 @@ var (
|
||||
)
|
||||
|
||||
func SchedulerWorkflow(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
|
||||
return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder())
|
||||
return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), false)
|
||||
}
|
||||
|
||||
func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.StartScheduleArgs, specBuilder *SpecBuilder) error {
|
||||
func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.StartScheduleArgs, specBuilder *SpecBuilder, enableCHASMMigration bool) error {
|
||||
scheduler := &scheduler{
|
||||
StartScheduleArgs: args,
|
||||
ctx: ctx,
|
||||
@@ -234,7 +235,8 @@ func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.St
|
||||
"namespace": args.State.Namespace,
|
||||
metrics.ScheduleBackendTag: metrics.ScheduleBackendLegacy,
|
||||
}),
|
||||
specBuilder: specBuilder,
|
||||
specBuilder: specBuilder,
|
||||
enableCHASMMigration: enableCHASMMigration,
|
||||
}
|
||||
return scheduler.run()
|
||||
}
|
||||
@@ -1261,7 +1263,12 @@ func (s *scheduler) checkConflict(token int64) error {
|
||||
|
||||
func (s *scheduler) updateTweakables() {
|
||||
// Use MutableSideEffect so that we can change the defaults without breaking determinism.
|
||||
get := func(ctx workflow.Context) any { return CurrentTweakablePolicies }
|
||||
enableCHASMMigration := s.enableCHASMMigration
|
||||
get := func(ctx workflow.Context) any {
|
||||
p := CurrentTweakablePolicies
|
||||
p.EnableCHASMMigration = enableCHASMMigration
|
||||
return p
|
||||
}
|
||||
eq := func(a, b any) bool { return a.(TweakablePolicies) == b.(TweakablePolicies) }
|
||||
if err := workflow.MutableSideEffect(s.ctx, "tweakables", get, eq).Get(&s.tweakables); err != nil {
|
||||
panic("can't decode TweakablePolicies:" + err.Error())
|
||||
|
||||
@@ -2466,17 +2466,18 @@ func (s *workflowSuite) TestMigrateFailureThenSignal() {
|
||||
}
|
||||
|
||||
func (s *workflowSuite) TestMigrateDynamicConfig() {
|
||||
// Enable migration via tweakable (simulating dynamic config).
|
||||
prevTweakables := CurrentTweakablePolicies
|
||||
CurrentTweakablePolicies.EnableCHASMMigration = true
|
||||
defer func() { CurrentTweakablePolicies = prevTweakables }()
|
||||
|
||||
// Enable migration by threading enableCHASMMigration=true through the closure (race-safe).
|
||||
// Mock MigrateSchedule activity to succeed.
|
||||
s.env.OnActivity(new(activities).MigrateScheduleToChasm, mock.Anything, mock.Anything).Once().Return(nil)
|
||||
|
||||
prevTweakables := CurrentTweakablePolicies
|
||||
CurrentTweakablePolicies.IterationsBeforeContinueAsNew = 100
|
||||
defer func() { CurrentTweakablePolicies = prevTweakables }()
|
||||
|
||||
s.env.SetStartTime(baseStartTime)
|
||||
s.env.ExecuteWorkflow(SchedulerWorkflow, &schedulespb.StartScheduleArgs{
|
||||
s.env.ExecuteWorkflow(func(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
|
||||
return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), true)
|
||||
}, &schedulespb.StartScheduleArgs{
|
||||
Schedule: &schedulepb.Schedule{
|
||||
Spec: &schedulepb.ScheduleSpec{
|
||||
Interval: []*schedulepb.IntervalSpec{{
|
||||
@@ -2499,11 +2500,8 @@ func (s *workflowSuite) TestMigrateDynamicConfig() {
|
||||
}
|
||||
|
||||
func (s *workflowSuite) TestMigrateDynamicConfigFailure() {
|
||||
// Enable migration via tweakable, but activity fails.
|
||||
prevTweakables := CurrentTweakablePolicies
|
||||
CurrentTweakablePolicies.EnableCHASMMigration = true
|
||||
defer func() { CurrentTweakablePolicies = prevTweakables }()
|
||||
|
||||
// Enable migration by threading enableCHASMMigration=true through the closure (race-safe),
|
||||
// but activity fails.
|
||||
migrateCalls := 0
|
||||
s.env.OnActivity(new(activities).MigrateScheduleToChasm, mock.Anything, mock.Anything).Return(
|
||||
func(context.Context, *schedulerpb.CreateFromMigrationStateRequest) error {
|
||||
@@ -2511,9 +2509,14 @@ func (s *workflowSuite) TestMigrateDynamicConfigFailure() {
|
||||
return errors.New("migration failed")
|
||||
})
|
||||
|
||||
prevTweakables := CurrentTweakablePolicies
|
||||
CurrentTweakablePolicies.IterationsBeforeContinueAsNew = 5
|
||||
defer func() { CurrentTweakablePolicies = prevTweakables }()
|
||||
|
||||
s.env.SetStartTime(baseStartTime)
|
||||
s.env.ExecuteWorkflow(SchedulerWorkflow, &schedulespb.StartScheduleArgs{
|
||||
s.env.ExecuteWorkflow(func(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
|
||||
return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), true)
|
||||
}, &schedulespb.StartScheduleArgs{
|
||||
Schedule: &schedulepb.Schedule{
|
||||
Spec: &schedulepb.ScheduleSpec{
|
||||
Interval: []*schedulepb.IntervalSpec{{
|
||||
|
||||
Reference in New Issue
Block a user