[Scheduler] Add a new flag to allow only migration of schedules without running workflows (#10505)

## What changed?
- Added a new dynamic config flag to control whether V1 schedules will
migrate with running workflows.

## Why?
- When V1 schedules migrate to V2 with running workflows, V2 attaches a
completion callback to the running workflow (as V2 exclusively uses
Nexus completion callbacks to monitor workflow status). Certain
third-party SDKs are known to have issues with one of the events written
to the history of the running workflow, which can cause them to panic
(Coinbase SDK). This allows us to dial up migration for these customers
without triggering that edge case.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [x] added new functional test(s)

## Potential risks
- Slower migration cadence, potentially

Author: lina-temporal

common/dynamicconfig/constants.go

@@ -3000,6 +3000,14 @@ EnableCHASMSchedulerMigration is true. The decision is re-evaluated when a
 scheduler workflow starts or continues-as-new.`,
 	)
 
+	EnableCHASMSchedulerMigrationWithRunningWorkflows = NewNamespaceBoolSetting(
+		"history.enableCHASMSchedulerMigrationWithRunningWorkflows",
+		false,
+		`EnableCHASMSchedulerMigrationWithRunningWorkflows, when set to false, prevents schedules with
+running workflows from being migrated. This works around a known bug in 3P SDKs involving updating
+existing workflows to attach callbacks.`,
+	)
+
 	EnableCHASMSchedulerSentinels = NewNamespaceBoolSetting(
 		"history.enableCHASMSchedulerSentinels",
 		true,

service/worker/scheduler/fx.go

@@ -54,6 +54,7 @@ type (
 		enabledForNs                 dynamicconfig.BoolPropertyFnWithNamespaceFilter
 		enableCHASMMigration         dynamicconfig.BoolPropertyFnWithNamespaceFilter
 		chasmMigrationRolloutPercent dynamicconfig.IntPropertyFnWithNamespaceFilter
+		migrateWithRunningWorkflows  dynamicconfig.BoolPropertyFnWithNamespaceFilter
 		globalNSStartWorkflowRPS     dynamicconfig.TypedSubscribableWithNamespaceFilter[float64]
 		maxBlobSize                  dynamicconfig.IntPropertyFnWithNamespaceFilter
 		localActivitySleepLimit      dynamicconfig.DurationPropertyFnWithNamespaceFilter
@@ -91,6 +92,7 @@ func NewResult(
 			enabledForNs:                 dynamicconfig.WorkerEnableScheduler.Get(dc),
 			enableCHASMMigration:         dynamicconfig.EnableCHASMSchedulerMigration.Get(dc),
 			chasmMigrationRolloutPercent: dynamicconfig.CHASMSchedulerMigrationRolloutPercent.Get(dc),
+			migrateWithRunningWorkflows:  dynamicconfig.EnableCHASMSchedulerMigrationWithRunningWorkflows.Get(dc),
 			globalNSStartWorkflowRPS:     dynamicconfig.SchedulerNamespaceStartWorkflowRPS.Subscribe(dc),
 			maxBlobSize:                  dynamicconfig.BlobSizeLimitError.Get(dc),
 			localActivitySleepLimit:      dynamicconfig.SchedulerLocalActivitySleepLimit.Get(dc),
@@ -112,7 +114,10 @@ func (s *workerComponent) Register(registry sdkworker.Registry, ns *namespace.Na
 			return s.enableCHASMMigration(nsName) &&
 				dynamicconfig.RolloutAccepts(key, s.chasmMigrationRolloutPercent(nsName))
 		}
-		return schedulerWorkflowWithSpecBuilder(ctx, args, s.specBuilder, enableMigration)
+		migrateWithRunningWorkflows := func() bool {
+			return s.migrateWithRunningWorkflows(nsName)
+		}
+		return schedulerWorkflowWithSpecBuilder(ctx, args, s.specBuilder, enableMigration, migrateWithRunningWorkflows)
 	}
 	registry.RegisterWorkflowWithOptions(wfFunc, workflow.RegisterOptions{Name: WorkflowType})
 

service/worker/scheduler/workflow.go

@@ -114,9 +114,10 @@ type (
 		// without modifying state).
 		specBuilder *SpecBuilder
 		cspec       *CompiledSpec
-		// enableCHASMMigration is re-evaluated every iteration inside the
-		// "tweakables" MutableSideEffect.
-		enableCHASMMigration func() bool
+		// enableCHASMMigration and migrateWithRunningWorkflows are re-evaluated
+		// every iteration inside the "tweakables" MutableSideEffect.
+		enableCHASMMigration        func() bool
+		migrateWithRunningWorkflows func() bool
 
 		tweakables TweakablePolicies
 
@@ -161,7 +162,8 @@ type (
 		Version                           SchedulerWorkflowVersion // Used to keep track of schedules version to release new features and for backward compatibility
 		// version 0 corresponds to the schedule version that comes before introducing the Version parameter
 
-		EnableCHASMMigration bool // Whether to automatically migrate this schedule to CHASM (V2)
+		EnableCHASMMigration        bool // Whether to automatically migrate this schedule to CHASM (V2)
+		MigrateWithRunningWorkflows bool // Whether to migrate this schedule to CHASM (V2) while it has running workflows
 
 		// When introducing a new field with new workflow logic, consider generating a new
 		// history for TestReplays using generate_history.sh.
@@ -226,10 +228,11 @@ var (
 )
 
 func SchedulerWorkflow(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
-	return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), func() bool { return false })
+	disabled := func() bool { return false }
+	return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), disabled, disabled)
 }
 
-func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.StartScheduleArgs, specBuilder *SpecBuilder, enableCHASMMigration func() bool) error {
+func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.StartScheduleArgs, specBuilder *SpecBuilder, enableCHASMMigration func() bool, migrateWithRunningWorkflows func() bool) error {
 	scheduler := &scheduler{
 		StartScheduleArgs: args,
 		ctx:               ctx,
@@ -239,8 +242,9 @@ func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.St
 			"namespace":                args.State.Namespace,
 			metrics.ScheduleBackendTag: metrics.ScheduleBackendLegacy,
 		}),
-		specBuilder:          specBuilder,
-		enableCHASMMigration: enableCHASMMigration,
+		specBuilder:                 specBuilder,
+		enableCHASMMigration:        enableCHASMMigration,
+		migrateWithRunningWorkflows: migrateWithRunningWorkflows,
 	}
 	return scheduler.run()
 }
@@ -324,7 +328,8 @@ func (s *scheduler) run() error {
 			)
 		}
 
-		if s.tweakables.EnableCHASMMigration {
+		if !s.State.PendingMigration && s.tweakables.EnableCHASMMigration &&
+			(s.tweakables.MigrateWithRunningWorkflows || len(s.Info.RunningWorkflows) == 0) {
 			s.State.PendingMigration = true
 		}
 		if s.State.PendingMigration {
@@ -1271,6 +1276,7 @@ func (s *scheduler) updateTweakables() {
 		p := CurrentTweakablePolicies
 		// Re-evaluates migration dynamic config each iteration.
 		p.EnableCHASMMigration = s.enableCHASMMigration()
+		p.MigrateWithRunningWorkflows = s.migrateWithRunningWorkflows()
 		return p
 	}
 	eq := func(a, b any) bool { return a.(TweakablePolicies) == b.(TweakablePolicies) }

service/worker/scheduler/workflow_test.go

@@ -2275,14 +2275,19 @@ func (s *workflowSuite) TestMigrateSuccess() {
 	// Mock MigrateSchedule activity to succeed.
 	s.env.OnActivity(new(activities).MigrateScheduleToChasm, mock.Anything, mock.Anything).Once().Return(nil)
 
-	// Send migrate signal after the first iteration.
+	// Enable migration and request it via signal after the first iteration.
+	enableMigration := false
 	s.env.RegisterDelayedCallback(func() {
+		enableMigration = true
 		s.env.SignalWorkflow(SignalNameMigrateToChasm, nil)
 	}, 1*time.Second)
 
 	CurrentTweakablePolicies.IterationsBeforeContinueAsNew = 100
 	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(),
+			func() bool { return enableMigration }, func() bool { return true })
+	}, &schedulespb.StartScheduleArgs{
 		Schedule: &schedulepb.Schedule{
 			Spec: &schedulepb.ScheduleSpec{
 				Interval: []*schedulepb.IntervalSpec{{
@@ -2314,8 +2319,10 @@ func (s *workflowSuite) TestMigrateFailure() {
 			return errors.New("migration failed")
 		})
 
-	// Send migrate signal after the first iteration.
+	// Enable migration and request it via signal after the first iteration.
+	enableMigration := false
 	s.env.RegisterDelayedCallback(func() {
+		enableMigration = true
 		s.env.SignalWorkflow(SignalNameMigrateToChasm, nil)
 	}, 1*time.Second)
 
@@ -2328,7 +2335,10 @@ func (s *workflowSuite) TestMigrateFailure() {
 
 	CurrentTweakablePolicies.IterationsBeforeContinueAsNew = 100
 	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(),
+			func() bool { return enableMigration }, func() bool { return true })
+	}, &schedulespb.StartScheduleArgs{
 		Schedule: &schedulepb.Schedule{
 			Spec: &schedulepb.ScheduleSpec{
 				Interval: []*schedulepb.IntervalSpec{{
@@ -2370,14 +2380,19 @@ func (s *workflowSuite) TestMigrateFailureThenRetrySuccess() {
 			return nil
 		})
 
-	// Send migrate signal after the first iteration.
+	// Enable migration and request it via signal after the first iteration.
+	enableMigration := false
 	s.env.RegisterDelayedCallback(func() {
+		enableMigration = true
 		s.env.SignalWorkflow(SignalNameMigrateToChasm, nil)
 	}, 1*time.Second)
 
 	CurrentTweakablePolicies.IterationsBeforeContinueAsNew = 100
 	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(),
+			func() bool { return enableMigration }, func() bool { return true })
+	}, &schedulespb.StartScheduleArgs{
 		Schedule: &schedulepb.Schedule{
 			Spec: &schedulepb.ScheduleSpec{
 				Interval: []*schedulepb.IntervalSpec{{
@@ -2410,8 +2425,10 @@ func (s *workflowSuite) TestMigrateFailureThenSignal() {
 			return errors.New("migration failed")
 		})
 
-	// Send migrate signal after the first iteration.
+	// Enable migration and request it via signal after the first iteration.
+	enableMigration := false
 	s.env.RegisterDelayedCallback(func() {
+		enableMigration = true
 		s.env.SignalWorkflow(SignalNameMigrateToChasm, nil)
 	}, 1*time.Second)
 	// After migration failure, send a pause patch and verify it's processed,
@@ -2434,7 +2451,10 @@ func (s *workflowSuite) TestMigrateFailureThenSignal() {
 
 	CurrentTweakablePolicies.IterationsBeforeContinueAsNew = 100
 	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(),
+			func() bool { return enableMigration }, func() bool { return true })
+	}, &schedulespb.StartScheduleArgs{
 		Schedule: &schedulepb.Schedule{
 			Spec: &schedulepb.ScheduleSpec{
 				Interval: []*schedulepb.IntervalSpec{{
@@ -2476,7 +2496,7 @@ func (s *workflowSuite) TestMigrateDynamicConfig() {
 
 	s.env.SetStartTime(baseStartTime)
 	s.env.ExecuteWorkflow(func(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
-		return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), func() bool { return true })
+		return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), func() bool { return true }, func() bool { return true })
 	}, &schedulespb.StartScheduleArgs{
 		Schedule: &schedulepb.Schedule{
 			Spec: &schedulepb.ScheduleSpec{
@@ -2525,7 +2545,7 @@ func (s *workflowSuite) TestMigrateDynamicConfigFlipsMidRun() {
 
 	s.env.SetStartTime(baseStartTime)
 	s.env.ExecuteWorkflow(func(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
-		return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), func() bool { return enabled })
+		return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), func() bool { return enabled }, func() bool { return true })
 	}, &schedulespb.StartScheduleArgs{
 		Schedule: &schedulepb.Schedule{
 			Spec: &schedulepb.ScheduleSpec{
@@ -2564,7 +2584,7 @@ func (s *workflowSuite) TestMigrateDynamicConfigFailure() {
 
 	s.env.SetStartTime(baseStartTime)
 	s.env.ExecuteWorkflow(func(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
-		return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), func() bool { return true })
+		return schedulerWorkflowWithSpecBuilder(ctx, args, NewSpecBuilder(), func() bool { return true }, func() bool { return true })
 	}, &schedulespb.StartScheduleArgs{
 		Schedule: &schedulepb.Schedule{
 			Spec: &schedulepb.ScheduleSpec{