Add intra-namespace percentage rollout for CHASM Nexus operation creation (#10950)

## What changed?
Adds a per-namespace **percentage** control for routing new
workflow-triggered Nexus operations to the CHASM implementation, on top
of the existing boolean `nexusoperation.enableChasmWorkflowOperations`
flag.

- New dynamic config
`nexusoperation.chasmWorkflowOperationsRolloutPercent` (int, 0–100),
**default 0**.
- The HSM→CHASM creation decision is centralized in one predicate,
`nexusoperation.UseChasmForWorkflow(enabled, rolloutPercent,
namespaceName, workflowID)`: an operation is created on the CHASM tree
only when the boolean flag is on **AND** the workflow falls within the
rollout percentage. Membership is decided by the shared `RolloutAccepts`
helper hashing `namespace + workflowID` (same key shape as the Scheduler
CHASM rollout), so a given workflow deterministically lands on the same
implementation across all of its operations and dialing the percentage
up is monotonic.
- **Both** framework-decision sites use the same predicate:
- **Live creation** — the CHASM `ScheduleNexusOperation` command handler
(`handleScheduleCommand`). Out-of-rollout → `ErrCommandNotSupported`, so
the operation is created on the HSM tree as before.
- **Reset / replication rebuild** —
`MutableStateRebuilder.applyChasmEvent`, on the
`NexusOperationScheduled` create event. Sharing the exact predicate is
required: otherwise a reset could flip an out-of-rollout workflow's
operations onto CHASM.
- **Cancelation is intentionally not gated** by this predicate. It is
routed by the tree that already owns the operation, so an operation
created while the flag/percentage was higher can still be canceled after
a downgrade.

## Why?
The per-namespace boolean is too coarse for a safe migration: turning it
on sends *all* of a namespace's new Nexus operations to CHASM at once. A
percentage control lets us dial CHASM adoption up gradually *within* a
namespace and roll back by dialing down — the de-risking control our
rollout plan requires. The default of 0 makes the boolean alone a no-op
until the percentage is explicitly raised, so enabling the flag can't
cause an all-at-once cutover.

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

## Potential risks
Low, and gated behind the default. Default 0 means behavior is unchanged
for any namespace that hasn't set the percentage — enabling the boolean
alone now routes nothing to CHASM until the percentage is raised. The
one behavioral consequence: any test/config that enables the boolean
expecting CHASM must also set the percentage (done for the tests in the
repo); no production config force-enables the boolean today.

Author: tekkaya

chasm/lib/nexusoperation/config.go

@@ -42,6 +42,29 @@ var EnableChasmWorkflowOperations = dynamicconfig.NewNamespaceBoolSetting(
 CHASM-based implementation of Nexus will be used when scheduling new Nexus Operations.`,
 )
 
+var ChasmWorkflowOperationsRolloutPercent = dynamicconfig.NewNamespaceIntSetting(
+	"nexusoperation.chasmWorkflowOperationsRolloutPercent",
+	0,
+	`Per-namespace percentage [0,100] of workflows whose new Nexus Operations are created on the CHASM-based
+implementation instead of the legacy HSM-based implementation. This setting is only consulted when
+enableChasmWorkflowOperations is true and is re-evaluated on every ScheduleNexusOperation command. Membership is
+decided by a stable hash of the namespace and workflow ID, so a given workflow consistently lands on the same
+implementation across all of its operations and dialing the percentage up is monotonic. Defaults to 0 (conservative
+dial-up model): even when enableChasmWorkflowOperations is on, no operations are routed to CHASM until the percentage
+is explicitly dialed up.`,
+)
+
+// UseChasmForWorkflow reports whether new Nexus operations for a workflow
+// should be created on CHASM. Live creation and rebuild must share this
+// predicate so reset keeps workflows in the same rollout bucket.
+func UseChasmForWorkflow(enabled bool, rolloutPercent int, namespaceName, workflowID string) bool {
+	if !enabled {
+		return false
+	}
+	key := fmt.Appendf(nil, "%s\x00%s", namespaceName, workflowID)
+	return dynamicconfig.RolloutAccepts(key, rolloutPercent)
+}
+
 var RequestTimeout = dynamicconfig.NewDestinationDurationSetting(
 	"nexusoperation.request.timeout",
 	time.Second*10,
@@ -218,40 +241,42 @@ Added for safety. Defaults to true. Likely to be removed in future server versio
 )
 
 type Config struct {
-	Enabled                             dynamicconfig.BoolPropertyFnWithNamespaceFilter
-	EnableChasm                         dynamicconfig.BoolPropertyFnWithNamespaceFilter
-	EnableChasmNexusWorkflowOperations  dynamicconfig.BoolPropertyFnWithNamespaceFilter
-	NumHistoryShards                    int32
-	LongPollBuffer                      dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	LongPollTimeout                     dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	RequestTimeout                      dynamicconfig.DurationPropertyFnWithDestinationFilter
-	MinRequestTimeout                   dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	MaxConcurrentOperationsPerWorkflow  dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxServiceNameLength                dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxOperationNameLength              dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxOperationTokenLength             dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxOperationHeaderSize              dynamicconfig.IntPropertyFnWithNamespaceFilter
-	DisallowedOperationHeaders          dynamicconfig.TypedPropertyFn[[]string]
-	MaxOperationScheduleToCloseTimeout  dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	PayloadSizeLimit                    dynamicconfig.IntPropertyFnWithNamespaceFilter
-	CallbackURLTemplate                 dynamicconfig.TypedPropertyFn[*template.Template]
-	UseSystemCallbackURL                dynamicconfig.BoolPropertyFn
-	PayloadSizeLimitWarn                dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxUserMetadataSummarySize          dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxUserMetadataDetailsSize          dynamicconfig.IntPropertyFnWithNamespaceFilter
-	UseNewFailureWireFormat             dynamicconfig.BoolPropertyFnWithNamespaceFilter
-	RecordCancelRequestCompletionEvents dynamicconfig.BoolPropertyFn
-	VisibilityMaxPageSize               dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxIDLengthLimit                    dynamicconfig.IntPropertyFn
-	MaxReasonLength                     dynamicconfig.IntPropertyFnWithNamespaceFilter
-	RetryPolicy                         func() backoff.RetryPolicy
+	Enabled                                    dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	EnableChasm                                dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	EnableChasmNexusWorkflowOperations         dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	ChasmNexusWorkflowOperationsRolloutPercent dynamicconfig.IntPropertyFnWithNamespaceFilter
+	NumHistoryShards                           int32
+	LongPollBuffer                             dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	LongPollTimeout                            dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	RequestTimeout                             dynamicconfig.DurationPropertyFnWithDestinationFilter
+	MinRequestTimeout                          dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	MaxConcurrentOperationsPerWorkflow         dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxServiceNameLength                       dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxOperationNameLength                     dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxOperationTokenLength                    dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxOperationHeaderSize                     dynamicconfig.IntPropertyFnWithNamespaceFilter
+	DisallowedOperationHeaders                 dynamicconfig.TypedPropertyFn[[]string]
+	MaxOperationScheduleToCloseTimeout         dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	PayloadSizeLimit                           dynamicconfig.IntPropertyFnWithNamespaceFilter
+	CallbackURLTemplate                        dynamicconfig.TypedPropertyFn[*template.Template]
+	UseSystemCallbackURL                       dynamicconfig.BoolPropertyFn
+	PayloadSizeLimitWarn                       dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxUserMetadataSummarySize                 dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxUserMetadataDetailsSize                 dynamicconfig.IntPropertyFnWithNamespaceFilter
+	UseNewFailureWireFormat                    dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	RecordCancelRequestCompletionEvents        dynamicconfig.BoolPropertyFn
+	VisibilityMaxPageSize                      dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxIDLengthLimit                           dynamicconfig.IntPropertyFn
+	MaxReasonLength                            dynamicconfig.IntPropertyFnWithNamespaceFilter
+	RetryPolicy                                func() backoff.RetryPolicy
 }
 
 func configProvider(dc *dynamicconfig.Collection, cfg *config.Persistence) *Config {
 	return &Config{
 		Enabled:                            Enabled.Get(dc),
 		EnableChasm:                        dynamicconfig.EnableChasm.Get(dc),
 		EnableChasmNexusWorkflowOperations: EnableChasmWorkflowOperations.Get(dc),
+		ChasmNexusWorkflowOperationsRolloutPercent: ChasmWorkflowOperationsRolloutPercent.Get(dc),
 		NumHistoryShards:                   cfg.NumHistoryShards,
 		LongPollBuffer:                     LongPollBuffer.Get(dc),
 		LongPollTimeout:                    LongPollTimeout.Get(dc),

chasm/lib/workflow/nexus_commands.go

@@ -35,7 +35,12 @@ func (ch *nexusCommandHandler) handleScheduleCommand(
 	ns := ctx.NamespaceEntry()
 	nsName := ns.Name().String()
 
-	if !ch.config.EnableChasmNexusWorkflowOperations(nsName) {
+	// Use the shared rollout predicate. If it rejects, let HSM create the operation.
+	if !nexusoperation.UseChasmForWorkflow(
+		ch.config.EnableChasmNexusWorkflowOperations(nsName),
+		ch.config.ChasmNexusWorkflowOperationsRolloutPercent(nsName),
+		nsName, ctx.ExecutionKey().BusinessID,
+	) {
 		return ErrCommandNotSupported
 	}
 

chasm/lib/workflow/nexus_commands_test.go

@@ -54,16 +54,21 @@ func (tcx *testContext) setHasAnyBufferedEvent(value bool) {
 }
 
 var defaultConfig = &nexusoperation.Config{
-	EnableChasmNexusWorkflowOperations: dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true),
-	MaxServiceNameLength:               dynamicconfig.GetIntPropertyFnFilteredByNamespace(len("service")),
-	MaxOperationNameLength:             dynamicconfig.GetIntPropertyFnFilteredByNamespace(len("op")),
-	MaxConcurrentOperationsPerWorkflow: dynamicconfig.GetIntPropertyFnFilteredByNamespace(2),
-	MaxOperationHeaderSize:             dynamicconfig.GetIntPropertyFnFilteredByNamespace(20),
-	DisallowedOperationHeaders:         dynamicconfig.GetTypedPropertyFn([]string{"request-timeout"}),
-	MaxOperationScheduleToCloseTimeout: dynamicconfig.GetDurationPropertyFnFilteredByNamespace(time.Hour * 24),
+	EnableChasmNexusWorkflowOperations:         dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true),
+	ChasmNexusWorkflowOperationsRolloutPercent: dynamicconfig.GetIntPropertyFnFilteredByNamespace(100),
+	MaxServiceNameLength:                       dynamicconfig.GetIntPropertyFnFilteredByNamespace(len("service")),
+	MaxOperationNameLength:                     dynamicconfig.GetIntPropertyFnFilteredByNamespace(len("op")),
+	MaxConcurrentOperationsPerWorkflow:         dynamicconfig.GetIntPropertyFnFilteredByNamespace(2),
+	MaxOperationHeaderSize:                     dynamicconfig.GetIntPropertyFnFilteredByNamespace(20),
+	DisallowedOperationHeaders:                 dynamicconfig.GetTypedPropertyFn([]string{"request-timeout"}),
+	MaxOperationScheduleToCloseTimeout:         dynamicconfig.GetDurationPropertyFnFilteredByNamespace(time.Hour * 24),
 }
 
-func newTestContext(t *testing.T, cfg *nexusoperation.Config) testContext {
+func newTestContext(t *testing.T, cfg *nexusoperation.Config, workflowID ...string) testContext {
+	wfID := ""
+	if len(workflowID) > 0 {
+		wfID = workflowID[0]
+	}
 	endpointReg := nexustest.FakeEndpointRegistry{
 		OnGetByName: func(ctx context.Context, namespaceID namespace.ID, endpointName string) (*persistencespb.NexusEndpointEntry, error) {
 			if endpointName == "endpoint caller namespace unauthorized" {
@@ -113,6 +118,7 @@ func newTestContext(t *testing.T, cfg *nexusoperation.Config) testContext {
 			HandleExecutionKey: func() chasm.ExecutionKey {
 				return chasm.ExecutionKey{
 					NamespaceID: tests.GlobalNamespaceEntry.ID().String(),
+					BusinessID:  wfID,
 				}
 			},
 			GoCtx: context.WithValue(context.Background(), nexusoperation.OperationContextKey, &nexusoperation.OperationContext{MetricTagConfig: dynamicconfig.GetTypedPropertyFn(nexusoperation.NexusMetricTagConfig{})}),
@@ -143,13 +149,63 @@ func newTestContext(t *testing.T, cfg *nexusoperation.Config) testContext {
 func TestHandleScheduleCommand(t *testing.T) {
 	t.Run("chasm nexus not enabled", func(t *testing.T) {
 		tcx := newTestContext(t, &nexusoperation.Config{
-			EnableChasmNexusWorkflowOperations: dynamicconfig.GetBoolPropertyFnFilteredByNamespace(false),
+			EnableChasmNexusWorkflowOperations:         dynamicconfig.GetBoolPropertyFnFilteredByNamespace(false),
+			ChasmNexusWorkflowOperationsRolloutPercent: dynamicconfig.GetIntPropertyFnFilteredByNamespace(100),
 		})
 		err := tcx.scheduleHandler(tcx.chasmCtx, tcx.wf, commandValidator{maxPayloadSize: 1}, &commandpb.Command{}, CommandHandlerOptions{WorkflowTaskCompletedEventID: 1})
 		require.ErrorIs(t, err, ErrCommandNotSupported)
 		require.Empty(t, tcx.history.Events)
 	})
 
+	validScheduleCmd := &commandpb.Command{
+		Attributes: &commandpb.Command_ScheduleNexusOperationCommandAttributes{
+			ScheduleNexusOperationCommandAttributes: &commandpb.ScheduleNexusOperationCommandAttributes{
+				Endpoint:  "endpoint",
+				Service:   "service",
+				Operation: "op",
+			},
+		},
+	}
+
+	t.Run("rollout percent 0 routes all operations to HSM", func(t *testing.T) {
+		cfg := *defaultConfig
+		cfg.ChasmNexusWorkflowOperationsRolloutPercent = dynamicconfig.GetIntPropertyFnFilteredByNamespace(0)
+		tcx := newTestContext(t, &cfg, "any-workflow")
+		err := tcx.scheduleHandler(tcx.chasmCtx, tcx.wf, commandValidator{maxPayloadSize: 1}, validScheduleCmd, CommandHandlerOptions{WorkflowTaskCompletedEventID: 1})
+		require.ErrorIs(t, err, ErrCommandNotSupported)
+		require.Empty(t, tcx.history.Events)
+	})
+
+	t.Run("rollout percent 100 routes all operations to CHASM", func(t *testing.T) {
+		cfg := *defaultConfig
+		cfg.ChasmNexusWorkflowOperationsRolloutPercent = dynamicconfig.GetIntPropertyFnFilteredByNamespace(100)
+		tcx := newTestContext(t, &cfg, "any-workflow")
+		err := tcx.scheduleHandler(tcx.chasmCtx, tcx.wf, commandValidator{maxPayloadSize: 1}, validScheduleCmd, CommandHandlerOptions{WorkflowTaskCompletedEventID: 1})
+		require.NoError(t, err)
+		require.Len(t, tcx.history.Events, 1)
+	})
+
+	t.Run("rollout percent routes by workflow id", func(t *testing.T) {
+		nsName := tests.GlobalNamespaceEntry.Name().String()
+		const percent = 50
+		// These IDs are pinned to opposite sides of the 50% rollout for this namespace.
+		const insideID, outsideID = "wf-4", "wf-0"
+		require.True(t, dynamicconfig.RolloutAccepts(fmt.Appendf(nil, "%s\x00%s", nsName, insideID), percent))
+		require.False(t, dynamicconfig.RolloutAccepts(fmt.Appendf(nil, "%s\x00%s", nsName, outsideID), percent))
+
+		cfg := *defaultConfig
+		cfg.ChasmNexusWorkflowOperationsRolloutPercent = dynamicconfig.GetIntPropertyFnFilteredByNamespace(percent)
+
+		scheduleFor := func(workflowID string) error {
+			tcx := newTestContext(t, &cfg, workflowID)
+			return tcx.scheduleHandler(tcx.chasmCtx, tcx.wf, commandValidator{maxPayloadSize: 1}, validScheduleCmd, CommandHandlerOptions{WorkflowTaskCompletedEventID: 1})
+		}
+
+		// A workflow inside the rollout routes to CHASM; one outside falls back to HSM.
+		require.NoError(t, scheduleFor(insideID))
+		require.ErrorIs(t, scheduleFor(outsideID), ErrCommandNotSupported)
+	})
+
 	t.Run("empty attributes", func(t *testing.T) {
 		tcx := newTestContext(t, defaultConfig)
 		err := tcx.scheduleHandler(tcx.chasmCtx, tcx.wf, commandValidator{maxPayloadSize: 1}, &commandpb.Command{}, CommandHandlerOptions{WorkflowTaskCompletedEventID: 1})