feat: [Scheduler] add worker.schedulerV1VersionOverride

Add a namespace dynamic config that advances the V1 scheduler workflow version without a server release. The override is reread every tweakables evaluation: it can advance the recorded version in the current run but never lowers it and never exceeds that run's recorded version ceiling. Values below the current default or above the latest supported version are ignored. - determineVersion / versionWithOverride apply the advancing-only override under the effective ceiling and the retained-version floor. - SchedulerV1VersionOverride setting + frontend Config and scheduler fx wiring; GetListInfoFromStartArgs threads the override for initial memo listing. - Override and rollback-ceiling unit coverage.
2026-08-30 18:41:49 -07:00 · 2026-08-27 17:36:38 -05:00
parent 7b78eb2c4d
commit c33dd5840c
7 changed files with 228 additions and 13 deletions
--- a/common/dynamicconfig/constants.go
+++ b/common/dynamicconfig/constants.go
@@ -3604,6 +3604,12 @@ is non-fatal: the search continues past this threshold.`,
 		`SchedulerV1VersionCeiling caps the workflow version the V1 scheduler records into history, so histories written on this cluster stay replayable on peer clusters that do not support newer versions. Set it to the highest scheduler version supported by the lowest peer. Intended for multi-cluster failover and rollback. The supported floor is version 1 (OSS v1.20). A negative value (the default) disables the cap.
 The ceiling is reread on every tweakables evaluation but only ratchets tighter within a run (one execution between start and Continue-As-New): a looser or unset value never raises it, and it never lowers a version already recorded for the run. Lowering the ceiling therefore takes effect on the next run, when Continue-As-New rereads dynamic config. A run is one execution between start and Continue-As-New.
 Operational notes: (1) A ceiling below 12 holds the version below CHASM migration support, so it pauses all V1->V2 CHASM migrations for the namespace until the ceiling is lifted (deferred, not dropped). (2) A ceiling below 6 skips custom search-attribute updates on schedule edits. (3) This caps V1 scheduler histories only; schedules already migrated to CHASM V2 are not made rollback-safe by it.`,
+	)
+	SchedulerV1VersionOverride = NewNamespaceIntSetting(
+		"worker.schedulerV1VersionOverride",
+		-1,
+		`SchedulerV1VersionOverride selects a newer V1 scheduler workflow version without requiring a follow-up server release to change the default. Set it to an explicitly supported version greater than or equal to the current default; a negative value (the default) keeps the current default. Values below the current default or above the latest version supported by this binary are ignored.
+The override is reread during every scheduler tweakables evaluation through MutableSideEffect. It can advance the version in the current workflow run at the next evaluation, but cannot lower it or exceed that run's recorded SchedulerV1VersionCeiling.`,
 	)
 	WorkerDeleteNamespaceActivityLimits = NewGlobalTypedSetting(
 		"worker.deleteNamespaceActivityLimitsConfig",
--- a/service/frontend/service.go
+++ b/service/frontend/service.go
@@ -175,6 +175,8 @@ type Config struct {
 	EnableSchedules dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	// Ceiling on the V1 scheduler workflow's recorded version.
 	SchedulerV1VersionCeiling dynamicconfig.IntPropertyFnWithNamespaceFilter
+	// Override for the V1 scheduler workflow's recorded version.
+	SchedulerV1VersionOverride dynamicconfig.IntPropertyFnWithNamespaceFilter

 	// Enable CHASM tree infrastructure
 	EnableChasm dynamicconfig.BoolPropertyFnWithNamespaceFilter
@@ -385,6 +387,7 @@ func NewConfig(

 		EnableSchedules:                      dynamicconfig.FrontendEnableSchedules.Get(dc),
 		SchedulerV1VersionCeiling:            dynamicconfig.SchedulerV1VersionCeiling.Get(dc),
+		SchedulerV1VersionOverride:           dynamicconfig.SchedulerV1VersionOverride.Get(dc),
 		EnableChasm:                          dynamicconfig.EnableChasm.Get(dc),
 		EnableCHASMSchedulerCreation:         dynamicconfig.EnableCHASMSchedulerCreation.Get(dc),
 		CHASMSchedulerCreationRolloutPercent: dynamicconfig.CHASMSchedulerCreationRolloutPercent.Get(dc),
--- a/service/frontend/workflow_handler.go
+++ b/service/frontend/workflow_handler.go
@@ -7068,7 +7068,8 @@ func (wh *WorkflowHandler) cleanScheduleMemo(memo *commonpb.Memo) *commonpb.Memo
 // This mutates request (but idempotent so safe for retries)
 func (wh *WorkflowHandler) addInitialScheduleMemo(request *workflowservice.CreateScheduleRequest, args *schedulespb.StartScheduleArgs) {
 	versionCeiling := wh.config.SchedulerV1VersionCeiling(request.Namespace)
-	info := scheduler.GetListInfoFromStartArgs(args, time.Now().UTC(), wh.scheduleSpecBuilder, versionCeiling)
+	versionOverride := wh.config.SchedulerV1VersionOverride(request.Namespace)
+	info := scheduler.GetListInfoFromStartArgs(args, time.Now().UTC(), wh.scheduleSpecBuilder, versionCeiling, versionOverride)
 	infoBytes, err := info.Marshal()
 	if err != nil {
 		wh.logger.Error("encoding initial schedule memo failed", tag.Error(err))
--- a/service/worker/scheduler/fx.go
+++ b/service/worker/scheduler/fx.go
@@ -56,6 +56,7 @@ type (
 		chasmMigrationRolloutPercent dynamicconfig.IntPropertyFnWithNamespaceFilter
 		migrateWithRunningWorkflows  dynamicconfig.BoolPropertyFnWithNamespaceFilter
 		schedulerV1VersionCeiling    dynamicconfig.IntPropertyFnWithNamespaceFilter
+		schedulerV1VersionOverride   dynamicconfig.IntPropertyFnWithNamespaceFilter
 		globalNSStartWorkflowRPS     dynamicconfig.TypedSubscribableWithNamespaceFilter[float64]
 		maxBlobSize                  dynamicconfig.IntPropertyFnWithNamespaceFilter
 		localActivitySleepLimit      dynamicconfig.DurationPropertyFnWithNamespaceFilter
@@ -95,6 +96,7 @@ func NewResult(
 			chasmMigrationRolloutPercent: dynamicconfig.CHASMSchedulerMigrationRolloutPercent.Get(dc),
 			migrateWithRunningWorkflows:  dynamicconfig.EnableCHASMSchedulerMigrationWithRunningWorkflows.Get(dc),
 			schedulerV1VersionCeiling:    dynamicconfig.SchedulerV1VersionCeiling.Get(dc),
+			schedulerV1VersionOverride:   dynamicconfig.SchedulerV1VersionOverride.Get(dc),
 			globalNSStartWorkflowRPS:     dynamicconfig.SchedulerNamespaceStartWorkflowRPS.Subscribe(dc),
 			maxBlobSize:                  dynamicconfig.BlobSizeLimitError.Get(dc),
 			localActivitySleepLimit:      dynamicconfig.SchedulerLocalActivitySleepLimit.Get(dc),
@@ -122,7 +124,10 @@ func (s *workerComponent) Register(registry sdkworker.Registry, ns *namespace.Na
 		versionCeiling := func() int {
 			return s.schedulerV1VersionCeiling(nsName)
 		}
-		return schedulerWorkflowWithSpecBuilder(ctx, args, s.specBuilder, enableMigration, migrateWithRunningWorkflows, versionCeiling)
+		versionOverride := func() int {
+			return s.schedulerV1VersionOverride(nsName)
+		}
+		return schedulerWorkflowWithSpecBuilderAndVersionOverride(ctx, args, s.specBuilder, enableMigration, migrateWithRunningWorkflows, versionCeiling, versionOverride)
 	}
 	registry.RegisterWorkflowWithOptions(wfFunc, workflow.RegisterOptions{Name: WorkflowType})

--- a/service/worker/scheduler/version_ceiling_test.go
+++ b/service/worker/scheduler/version_ceiling_test.go
@@ -32,6 +32,26 @@ func TestClampVersion(t *testing.T) {
 	}
 }

+func TestVersionWithOverride(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		ceiling  int
+		override int
+		want     SchedulerWorkflowVersion
+	}{
+		{name: "default version", ceiling: -1, override: -1, want: TriggerImmediatelyTimestamp},
+		{name: "override latest", ceiling: -1, override: int(LatestSchedulerWorkflowVersion), want: LatestSchedulerWorkflowVersion},
+		{name: "ceiling caps override", ceiling: int(TriggerImmediatelyTimestamp) - 1, override: int(LatestSchedulerWorkflowVersion), want: TriggerImmediatelyTimestamp - 1},
+		{name: "override below current ignored", ceiling: -1, override: int(TriggerImmediatelyTimestamp) - 1, want: TriggerImmediatelyTimestamp},
+		{name: "override above latest ignored", ceiling: -1, override: int(LatestSchedulerWorkflowVersion) + 1, want: TriggerImmediatelyTimestamp},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			version := versionWithOverride(TriggerImmediatelyTimestamp, tc.ceiling, tc.override)
+			require.Equal(t, tc.want, version)
+		})
+	}
+}
+
 func TestDetermineVersionTransitions(t *testing.T) {
 	for _, tc := range []struct {
 		name     string
@@ -83,6 +103,81 @@ func TestDetermineVersionTransitions(t *testing.T) {
 	}
 }

+// TestDetermineVersionTightenOnly drives multi-iteration runs where the configured ceiling and
+// override change between iterations, asserting the two run-scoped invariants: the effective
+// ceiling only ratchets tighter, and the version never decreases within a run (a lowered ceiling
+// is retained as a recorded floor and only downgrades on the next run).
+func TestDetermineVersionTightenOnly(t *testing.T) {
+	const v12 = TriggerImmediatelyTimestamp
+	const v13 = MigrationHandoffFixes
+	type step struct {
+		def      SchedulerWorkflowVersion
+		ceiling  int
+		override int
+		wantVer  SchedulerWorkflowVersion
+		wantCeil int
+	}
+	for _, tc := range []struct {
+		name  string
+		steps []step
+	}{
+		{
+			name: "tighten to 12 blocks a simultaneous override to 13",
+			steps: []step{
+				{def: v12, ceiling: -1, override: -1, wantVer: v12, wantCeil: -1},
+				{def: v12, ceiling: int(v12), override: int(v13), wantVer: v12, wantCeil: int(v12)},
+			},
+		},
+		{
+			name: "default advance to 13 is blocked by a tightened ceiling",
+			steps: []step{
+				{def: v12, ceiling: int(v12), override: -1, wantVer: v12, wantCeil: int(v12)},
+				{def: v13, ceiling: int(v12), override: -1, wantVer: v12, wantCeil: int(v12)},
+			},
+		},
+		{
+			name: "ceiling cannot loosen to 13 or unset within a run",
+			steps: []step{
+				{def: v12, ceiling: int(v12), override: -1, wantVer: v12, wantCeil: int(v12)},
+				{def: v13, ceiling: int(v13), override: int(v13), wantVer: v12, wantCeil: int(v12)},
+				{def: v13, ceiling: -1, override: int(v13), wantVer: v12, wantCeil: int(v12)},
+			},
+		},
+		{
+			name: "tightening below a recorded v13 keeps the version and records the tighter ceiling",
+			steps: []step{
+				{def: v12, ceiling: -1, override: int(v13), wantVer: v13, wantCeil: -1},
+				{def: v12, ceiling: int(v12), override: int(v13), wantVer: v13, wantCeil: int(v12)},
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var ceiling, override int
+			s := &scheduler{
+				logger:          log.NewSdkLogger(log.NewNoopLogger()),
+				versionCeiling:  func() int { return ceiling },
+				versionOverride: func() int { return override },
+			}
+			for i, st := range tc.steps {
+				ceiling, override = st.ceiling, st.override
+				version, recordedCeiling := s.determineVersion(st.def)
+				require.Equalf(t, st.wantVer, version, "step %d version", i)
+				require.Equalf(t, st.wantCeil, recordedCeiling, "step %d recorded ceiling", i)
+				// A version above the effective ceiling is only ever the retained recorded floor.
+				if recordedCeiling >= 0 && version > SchedulerWorkflowVersion(recordedCeiling) {
+					require.GreaterOrEqualf(t, s.tweakables.Version, version,
+						"step %d: version above ceiling must be the retained recorded floor", i)
+				}
+				// MutableSideEffect stores this value for the next workflow task.
+				s.tweakables = CurrentTweakablePolicies
+				s.tweakables.Version = version
+				s.tweakables.VersionCeiling = recordedCeiling
+				s.tweakables.VersionCeilingSet = true
+			}
+		})
+	}
+}
+
 // TestDetermineVersionDowngradeOnNextRun verifies that after a run kept v13 as a floor while the
 // ceiling was tightened to 12, the following run (fresh tweakables after continue-as-new) reads
 // dynamic config and starts capped at 12.
@@ -148,6 +243,51 @@ func TestDetermineVersionPreservesLegacyRecordedVersion(t *testing.T) {
 	}
 }

+func TestDetermineVersionAdvancesWithOverride(t *testing.T) {
+	override := -1
+	s := &scheduler{
+		logger:          log.NewSdkLogger(log.NewNoopLogger()),
+		versionCeiling:  func() int { return -1 },
+		versionOverride: func() int { return override },
+	}
+
+	version, ceiling := s.determineVersion(TriggerImmediatelyTimestamp)
+	require.Equal(t, SchedulerWorkflowVersion(TriggerImmediatelyTimestamp), version)
+	require.Equal(t, -1, ceiling)
+	s.tweakables = CurrentTweakablePolicies
+	s.tweakables.Version = version
+	s.tweakables.VersionCeiling = ceiling
+	s.tweakables.VersionCeilingSet = true
+
+	override = int(LatestSchedulerWorkflowVersion)
+	version, ceiling = s.determineVersion(TriggerImmediatelyTimestamp)
+	require.Equal(t, SchedulerWorkflowVersion(LatestSchedulerWorkflowVersion), version)
+	require.Equal(t, -1, ceiling)
+	s.tweakables.Version = version
+
+	override = -1
+	version, ceiling = s.determineVersion(TriggerImmediatelyTimestamp)
+	require.Equal(t, SchedulerWorkflowVersion(LatestSchedulerWorkflowVersion), version)
+	require.Equal(t, -1, ceiling)
+}
+
+func TestDetermineVersionOverrideRespectsRecordedCeiling(t *testing.T) {
+	s := &scheduler{
+		logger:          log.NewSdkLogger(log.NewNoopLogger()),
+		versionCeiling:  func() int { return oldPeerCeiling },
+		versionOverride: func() int { return int(LatestSchedulerWorkflowVersion) },
+		tweakables: TweakablePolicies{
+			Version:           oldPeerCeiling,
+			VersionCeiling:    oldPeerCeiling,
+			VersionCeilingSet: true,
+		},
+	}
+
+	version, ceiling := s.determineVersion(TriggerImmediatelyTimestamp)
+	require.Equal(t, SchedulerWorkflowVersion(oldPeerCeiling), version)
+	require.Equal(t, int(oldPeerCeiling), ceiling)
+}
+
 // TestVersionCeilingWithCHASMMigration verifies that a clamp below the CHASM gate keeps migration
 // markers out of history, and that once the ceiling is lifted (on the next run) the deferred
 // migration runs.
--- a/service/worker/scheduler/workflow.go
+++ b/service/worker/scheduler/workflow.go
@@ -131,11 +131,13 @@ type (
 		// inside the "tweakables" MutableSideEffect.
 		enableCHASMMigration        func() bool
 		migrateWithRunningWorkflows func() bool
-		// versionCeiling is re-evaluated every iteration inside the "tweakables" MutableSideEffect,
-		// alongside the migration knobs above. The ceiling only ratchets tighter within a run (a
-		// looser or unset value never raises it) and never lowers an already-recorded version
-		// mid-run -- a lowered ceiling takes effect on the next run. See determineVersion.
-		versionCeiling func() int
+		// versionCeiling and versionOverride are re-evaluated every iteration inside the
+		// "tweakables" MutableSideEffect, alongside the migration knobs above. The ceiling only
+		// ratchets tighter within a run (a looser or unset value never raises it); the override
+		// only advances the version. Neither can lower an already-recorded version mid-run --
+		// a lowered ceiling takes effect on the next run. See determineVersion.
+		versionCeiling  func() int
+		versionOverride func() int

 		tweakables TweakablePolicies

@@ -258,6 +260,10 @@ func SchedulerWorkflow(ctx workflow.Context, args *schedulespb.StartScheduleArgs
 }

 func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.StartScheduleArgs, specBuilder *SpecBuilder, enableCHASMMigration func() bool, migrateWithRunningWorkflows func() bool, versionCeiling func() int) error {
+	return schedulerWorkflowWithSpecBuilderAndVersionOverride(ctx, args, specBuilder, enableCHASMMigration, migrateWithRunningWorkflows, versionCeiling, func() int { return -1 })
+}
+
+func schedulerWorkflowWithSpecBuilderAndVersionOverride(ctx workflow.Context, args *schedulespb.StartScheduleArgs, specBuilder *SpecBuilder, enableCHASMMigration func() bool, migrateWithRunningWorkflows func() bool, versionCeiling func() int, versionOverride func() int) error {
 	scheduler := &scheduler{
 		StartScheduleArgs: args,
 		ctx:               ctx,
@@ -271,6 +277,7 @@ func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.St
 		enableCHASMMigration:        enableCHASMMigration,
 		migrateWithRunningWorkflows: migrateWithRunningWorkflows,
 		versionCeiling:              versionCeiling,
+		versionOverride:             versionOverride,
 	}
 	return scheduler.run()
 }
@@ -1839,11 +1846,20 @@ func (s *scheduler) determineVersion(defaultVersion SchedulerWorkflowVersion) (S
 	}
 	ceiling := tightenCeiling(recordedCeiling, configured)

-	// Never regress below the version already recorded for this run. Apply the effective ceiling,
-	// then re-floor at the recorded version so a lowered ceiling cannot downgrade a running
-	// workflow mid-run.
+	override := -1
+	if s.versionOverride != nil {
+		override = s.versionOverride()
+	}
+	if override > int(LatestSchedulerWorkflowVersion) {
+		s.logger.Warn("worker.schedulerV1VersionOverride above the latest supported version; ignored",
+			"override", override, "latestSupportedVersion", LatestSchedulerWorkflowVersion)
+	}
+
+	// Never regress below the version already recorded for this run. Apply the advancing-only
+	// override and the effective ceiling, then re-floor at the recorded version so a lowered
+	// ceiling cannot downgrade a running workflow mid-run.
 	candidate := max(defaultVersion, s.tweakables.Version)
-	version := max(clampVersion(candidate, ceiling), s.tweakables.Version)
+	version := max(versionWithOverride(candidate, ceiling, override), s.tweakables.Version)
 	return version, ceiling
 }

@@ -1877,11 +1893,11 @@ func panicIfErr(err error) {
 	}
 }

-func GetListInfoFromStartArgs(args *schedulespb.StartScheduleArgs, now time.Time, specBuilder *SpecBuilder, versionCeiling int) *schedulepb.ScheduleListInfo {
+func GetListInfoFromStartArgs(args *schedulespb.StartScheduleArgs, now time.Time, specBuilder *SpecBuilder, versionCeiling, versionOverride int) *schedulepb.ScheduleListInfo {
 	// Create a scheduler outside of workflow context with just the fields we need to call
 	// getListInfo. Note that this does not take into account InitialPatch.
 	tweakables := CurrentTweakablePolicies
-	tweakables.Version = clampVersion(tweakables.Version, versionCeiling)
+	tweakables.Version = versionWithOverride(tweakables.Version, versionCeiling, versionOverride)
 	s := &scheduler{
 		StartScheduleArgs: args,
 		tweakables:        tweakables,
@@ -1893,6 +1909,13 @@ func GetListInfoFromStartArgs(args *schedulespb.StartScheduleArgs, now time.Time
 	return s.getListInfo(false)
 }

+func versionWithOverride(defaultVersion SchedulerWorkflowVersion, ceiling, override int) SchedulerWorkflowVersion {
+	if override >= int(defaultVersion) && override <= int(LatestSchedulerWorkflowVersion) {
+		defaultVersion = SchedulerWorkflowVersion(override)
+	}
+	return clampVersion(defaultVersion, ceiling)
+}
+
 func isUserScheduleError(err error) bool {
 	var appError *temporal.ApplicationError
 	if errors.As(err, &appError) && appError.Type() == workflowExecutionAlreadyStarted {
--- a/service/worker/scheduler/workflow_test.go
+++ b/service/worker/scheduler/workflow_test.go
@@ -401,6 +401,43 @@ func (s *workflowSuite) TestMigratedBufferedStartUsesLegacyIDsAtOldVersion() {
 	s.True(workflow.IsContinueAsNewError(s.env.GetWorkflowError()))
 }

+// TestMigratedBufferedStartLosesIdempotencyIDsUnderCeiling characterizes a known constraint of the
+// version ceiling: the v13 (MigrationHandoffFixes) preservation of a migrated start's
+// workflow/request IDs is undone if the recorded version is clamped below v13. This is the
+// failover path -- default is v13 but a rollback ceiling of v12 holds the run at v12 -- so a start
+// that was migrated from CHASM with preserved idempotency IDs is started with a freshly generated
+// request ID and the legacy generated workflow ID instead. Lowering the ceiling mid-flight can
+// therefore break migrated-start idempotency; documented rather than guarded (guarding would mean
+// refusing to clamp below the version that produced pending buffered starts).
+func (s *workflowSuite) TestMigratedBufferedStartLosesIdempotencyIDsUnderCeiling() {
+	previousTweakables := CurrentTweakablePolicies
+	defer func() { CurrentTweakablePolicies = previousTweakables }()
+	// Default is the newest version; only the ceiling holds the run down to v12.
+	CurrentTweakablePolicies.Version = MigrationHandoffFixes
+	CurrentTweakablePolicies.IterationsBeforeContinueAsNew = 1
+
+	s.expectStart(func(req *schedulespb.StartWorkflowRequest) (*schedulespb.StartWorkflowResponse, error) {
+		s.Equal("configured-workflow-id-2022-06-01T00:00:00Z", req.Request.WorkflowId)
+		s.NotEmpty(req.Request.RequestId)
+		s.NotEqual("migrated-request-id", req.Request.RequestId)
+		return nil, nil
+	})
+
+	wf := func(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error {
+		return schedulerWorkflowWithSpecBuilderAndVersionOverride(
+			ctx, args, newSpecBuilderForTest(0, 0),
+			func() bool { return false },                           // enableCHASMMigration
+			func() bool { return false },                           // migrateWithRunningWorkflows
+			func() int { return int(TriggerImmediatelyTimestamp) }, // versionCeiling: rollback clamp below v13
+			func() int { return -1 },                               // versionOverride
+		)
+	}
+	s.env.SetStartTime(baseStartTime)
+	s.env.ExecuteWorkflow(wf, s.migratedStartScheduleArgs())
+	s.True(s.env.IsWorkflowCompleted())
+	s.True(workflow.IsContinueAsNewError(s.env.GetWorkflowError()))
+}
+
 func (s *workflowSuite) TestNativeBufferedStartFallsBackAtNewVersion() {
 	// Complement to TestMigratedBufferedStart*: a schedule that never went through
 	// CHASM has BufferedStart.WorkflowId/RequestId empty (the common case -- see