add more unittests and functional tests

Author: feiyang3cat

service/frontend/workflow_handler.go

@@ -2455,6 +2455,17 @@ func (wh *WorkflowHandler) ResetWorkflowExecution(ctx context.Context, request *
 		return nil, serviceerror.NewInternalf("unknown reset reapply type: %v", request.GetResetReapplyType())
 	}
 
+	for _, postOp := range request.GetPostResetOperations() {
+		if updateOpts := postOp.GetUpdateWorkflowOptions(); updateOpts != nil {
+			if err := wh.validateTimeSkippingConfig(
+				updateOpts.GetWorkflowExecutionOptions().GetTimeSkippingConfig(),
+				namespace.Name(request.GetNamespace()),
+			); err != nil {
+				return nil, err
+			}
+		}
+	}
+
 	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespace.Name(request.GetNamespace()))
 	if err != nil {
 		return nil, err
@@ -5668,9 +5679,25 @@ func (wh *WorkflowHandler) StartBatchOperation(
 	case *workflowservice.StartBatchOperationRequest_ResetOperation:
 		input.BatchType = enumspb.BATCH_OPERATION_TYPE_RESET
 		identity = op.ResetOperation.GetIdentity()
+		for _, postOp := range op.ResetOperation.GetPostResetOperations() {
+			if updateOpts := postOp.GetUpdateWorkflowOptions(); updateOpts != nil {
+				if err := wh.validateTimeSkippingConfig(
+					updateOpts.GetWorkflowExecutionOptions().GetTimeSkippingConfig(),
+					namespace.Name(request.GetNamespace()),
+				); err != nil {
+					return nil, err
+				}
+			}
+		}
 	case *workflowservice.StartBatchOperationRequest_UpdateWorkflowOptionsOperation:
 		input.BatchType = enumspb.BATCH_OPERATION_TYPE_UPDATE_EXECUTION_OPTIONS
 		identity = op.UpdateWorkflowOptionsOperation.GetIdentity()
+		if err := wh.validateTimeSkippingConfig(
+			op.UpdateWorkflowOptionsOperation.GetWorkflowExecutionOptions().GetTimeSkippingConfig(),
+			namespace.Name(request.GetNamespace()),
+		); err != nil {
+			return nil, err
+		}
 	case *workflowservice.StartBatchOperationRequest_UnpauseActivitiesOperation:
 		input.BatchType = enumspb.BATCH_OPERATION_TYPE_UNPAUSE_ACTIVITY
 		identity = op.UnpauseActivitiesOperation.GetIdentity()

service/frontend/workflow_handler_test.go

@@ -3486,6 +3486,107 @@ func (s *WorkflowHandlerSuite) TestSignalWithStartWorkflowExecution_TimeSkipping
 	s.True(resp.Started)
 }
 
+// TestResetWorkflowExecution_TimeSkipping_DCDisabled verifies that when the DC gate is off,
+// a ResetWorkflowExecution request with a TimeSkippingConfig inside PostResetOperations is rejected.
+func (s *WorkflowHandlerSuite) TestResetWorkflowExecution_TimeSkipping_DCDisabled() {
+	config := s.newConfig()
+	config.TimeSkippingEnabled = dc.GetBoolPropertyFnFilteredByNamespace(false)
+	wh := s.getWorkflowHandler(config)
+
+	_, err := wh.ResetWorkflowExecution(context.Background(), &workflowservice.ResetWorkflowExecutionRequest{
+		Namespace: s.testNamespace.String(),
+		RequestId: uuid.NewString(),
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: "workflow-id",
+			RunId:      uuid.NewString(),
+		},
+		PostResetOperations: []*workflowpb.PostResetOperation{
+			{
+				Variant: &workflowpb.PostResetOperation_UpdateWorkflowOptions_{
+					UpdateWorkflowOptions: &workflowpb.PostResetOperation_UpdateWorkflowOptions{
+						WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{
+							TimeSkippingConfig: &workflowpb.TimeSkippingConfig{Enabled: true},
+						},
+						UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}},
+					},
+				},
+			},
+		},
+	})
+	var unimplemented *serviceerror.Unimplemented
+	s.ErrorAs(err, &unimplemented)
+	s.ErrorContains(err, "The Time-Skipping feature is not enabled for namespace")
+}
+
+// TestStartBatchOperation_ResetOperation_TimeSkipping_DCDisabled verifies that when the DC gate
+// is off, a batch reset with a TimeSkippingConfig inside PostResetOperations is rejected.
+func (s *WorkflowHandlerSuite) TestStartBatchOperation_ResetOperation_TimeSkipping_DCDisabled() {
+	config := s.newConfig()
+	config.TimeSkippingEnabled = dc.GetBoolPropertyFnFilteredByNamespace(false)
+	wh := s.getWorkflowHandler(config)
+	namespaceID := namespace.ID(uuid.NewString())
+	s.mockNamespaceCache.EXPECT().GetNamespaceID(gomock.Any()).Return(namespaceID, nil).AnyTimes()
+	s.mockVisibilityMgr.EXPECT().CountWorkflowExecutions(gomock.Any(), gomock.Any()).Return(&manager.CountWorkflowExecutionsResponse{Count: 0}, nil)
+
+	_, err := wh.StartBatchOperation(context.Background(), &workflowservice.StartBatchOperationRequest{
+		Namespace:       s.testNamespace.String(),
+		JobId:           uuid.NewString(),
+		Reason:          "test",
+		VisibilityQuery: "WorkflowId='test'",
+		Operation: &workflowservice.StartBatchOperationRequest_ResetOperation{
+			ResetOperation: &batchpb.BatchOperationReset{
+				Options: &commonpb.ResetOptions{
+					Target: &commonpb.ResetOptions_WorkflowTaskId{WorkflowTaskId: 1},
+				},
+				PostResetOperations: []*workflowpb.PostResetOperation{
+					{
+						Variant: &workflowpb.PostResetOperation_UpdateWorkflowOptions_{
+							UpdateWorkflowOptions: &workflowpb.PostResetOperation_UpdateWorkflowOptions{
+								WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{
+									TimeSkippingConfig: &workflowpb.TimeSkippingConfig{Enabled: true},
+								},
+								UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}},
+							},
+						},
+					},
+				},
+			},
+		},
+	})
+	var unimplemented *serviceerror.Unimplemented
+	s.ErrorAs(err, &unimplemented)
+	s.ErrorContains(err, "The Time-Skipping feature is not enabled for namespace")
+}
+
+// TestStartBatchOperation_UpdateWorkflowOptionsOperation_TimeSkipping_DCDisabled verifies that
+// when the DC gate is off, a batch UpdateWorkflowOptions with a TimeSkippingConfig is rejected.
+func (s *WorkflowHandlerSuite) TestStartBatchOperation_UpdateWorkflowOptionsOperation_TimeSkipping_DCDisabled() {
+	config := s.newConfig()
+	config.TimeSkippingEnabled = dc.GetBoolPropertyFnFilteredByNamespace(false)
+	wh := s.getWorkflowHandler(config)
+	namespaceID := namespace.ID(uuid.NewString())
+	s.mockNamespaceCache.EXPECT().GetNamespaceID(gomock.Any()).Return(namespaceID, nil).AnyTimes()
+	s.mockVisibilityMgr.EXPECT().CountWorkflowExecutions(gomock.Any(), gomock.Any()).Return(&manager.CountWorkflowExecutionsResponse{Count: 0}, nil)
+
+	_, err := wh.StartBatchOperation(context.Background(), &workflowservice.StartBatchOperationRequest{
+		Namespace:       s.testNamespace.String(),
+		JobId:           uuid.NewString(),
+		Reason:          "test",
+		VisibilityQuery: "WorkflowId='test'",
+		Operation: &workflowservice.StartBatchOperationRequest_UpdateWorkflowOptionsOperation{
+			UpdateWorkflowOptionsOperation: &batchpb.BatchOperationUpdateWorkflowExecutionOptions{
+				WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{
+					TimeSkippingConfig: &workflowpb.TimeSkippingConfig{Enabled: true},
+				},
+				UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}},
+			},
+		},
+	})
+	var unimplemented *serviceerror.Unimplemented
+	s.ErrorAs(err, &unimplemented)
+	s.ErrorContains(err, "The Time-Skipping feature is not enabled for namespace")
+}
+
 func (s *WorkflowHandlerSuite) newConfig() *Config {
 	return NewConfig(dc.NewNoopCollection(), numHistoryShards)
 }
@@ -4475,6 +4576,29 @@ func (s *WorkflowHandlerSuite) TestUpdateWorkflowExecutionOptions_Priority() {
 	// NOTE: only testing a single validation scenario here; the priority validation has its own unit tests
 }
 
+// TestUpdateWorkflowExecutionOptions_TimeSkipping_DCDisabled verifies that when the DC gate is
+// off, UpdateWorkflowExecutionOptions rejects a request containing a TimeSkippingConfig.
+func (s *WorkflowHandlerSuite) TestUpdateWorkflowExecutionOptions_TimeSkipping_DCDisabled() {
+	config := s.newConfig()
+	config.TimeSkippingEnabled = dc.GetBoolPropertyFnFilteredByNamespace(false)
+	wh := s.getWorkflowHandler(config)
+
+	_, err := wh.UpdateWorkflowExecutionOptions(context.Background(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{
+		Namespace: s.testNamespace.String(),
+		WorkflowExecution: &commonpb.WorkflowExecution{
+			WorkflowId: "workflow-id",
+			RunId:      "run-id",
+		},
+		WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{
+			TimeSkippingConfig: &workflowpb.TimeSkippingConfig{Enabled: true},
+		},
+		UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}},
+	})
+	var unimplemented *serviceerror.Unimplemented
+	s.ErrorAs(err, &unimplemented)
+	s.ErrorContains(err, "The Time-Skipping feature is not enabled for namespace")
+}
+
 func (s *WorkflowHandlerSuite) TestUpdateActivityOptions_Priority() {
 	config := s.newConfig()
 	wh := s.getWorkflowHandler(config)

service/history/api/updateworkflowoptions/api_test.go

@@ -3,8 +3,8 @@ package updateworkflowoptions
 import (
 	"context"
 	"testing"
+	"time"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	commonpb "go.temporal.io/api/common/v1"
@@ -28,6 +28,7 @@ import (
 	wcache "go.temporal.io/server/service/history/workflow/cache"
 	"go.uber.org/mock/gomock"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/fieldmaskpb"
 )
 
@@ -94,28 +95,28 @@ func TestMergeOptions_VersionOverrideMask(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-	assert.EqualExportedValues(t, unpinnedOverrideOptions, merged)
+	require.EqualExportedValues(t, unpinnedOverrideOptions, merged)
 
 	// Merge pinned_A into unpinned options
 	merged, err = mergeWorkflowExecutionOptions(input, pinnedOverrideOptionsA, updateMask)
 	if err != nil {
 		t.Error(err)
 	}
-	assert.EqualExportedValues(t, pinnedOverrideOptionsA, merged)
+	require.EqualExportedValues(t, pinnedOverrideOptionsA, merged)
 
 	// Merge pinned_B into pinned_A options
 	merged, err = mergeWorkflowExecutionOptions(input, pinnedOverrideOptionsB, updateMask)
 	if err != nil {
 		t.Error(err)
 	}
-	assert.EqualExportedValues(t, pinnedOverrideOptionsB, merged)
+	require.EqualExportedValues(t, pinnedOverrideOptionsB, merged)
 
 	// Unset versioning override
 	merged, err = mergeWorkflowExecutionOptions(input, emptyOptions, updateMask)
 	if err != nil {
 		t.Error(err)
 	}
-	assert.EqualExportedValues(t, emptyOptions, merged)
+	require.EqualExportedValues(t, emptyOptions, merged)
 }
 
 func TestMergeOptions_PartialMask(t *testing.T) {
@@ -124,14 +125,14 @@ func TestMergeOptions_PartialMask(t *testing.T) {
 	deploymentOnlyUpdateMask := &fieldmaskpb.FieldMask{Paths: []string{"versioning_override.deployment"}}
 
 	_, err := mergeWorkflowExecutionOptions(emptyOptions, unpinnedOverrideOptions, behaviorOnlyUpdateMask)
-	assert.Error(t, err)
+	require.Error(t, err)
 
 	_, err = mergeWorkflowExecutionOptions(emptyOptions, unpinnedOverrideOptions, deploymentOnlyUpdateMask)
-	assert.Error(t, err)
+	require.Error(t, err)
 
 	merged, err := mergeWorkflowExecutionOptions(emptyOptions, unpinnedOverrideOptions, bothUpdateMask)
-	assert.NoError(t, err)
-	assert.EqualExportedValues(t, unpinnedOverrideOptions, merged)
+	require.NoError(t, err)
+	require.EqualExportedValues(t, unpinnedOverrideOptions, merged)
 }
 
 func TestMergeOptions_EmptyMask(t *testing.T) {
@@ -140,25 +141,113 @@ func TestMergeOptions_EmptyMask(t *testing.T) {
 
 	// Don't merge anything
 	merged, err := mergeWorkflowExecutionOptions(input, pinnedOverrideOptionsA, emptyUpdateMask)
-	assert.NoError(t, err)
-	assert.EqualExportedValues(t, input, merged)
+	require.NoError(t, err)
+	require.EqualExportedValues(t, input, merged)
 
 	// Don't merge anything
 	merged, err = mergeWorkflowExecutionOptions(input, nil, emptyUpdateMask)
-	assert.NoError(t, err)
-	assert.EqualExportedValues(t, input, merged)
+	require.NoError(t, err)
+	require.EqualExportedValues(t, input, merged)
 }
 
 func TestMergeOptions_AsteriskMask(t *testing.T) {
 	asteriskUpdateMask := &fieldmaskpb.FieldMask{Paths: []string{"*"}}
 	_, err := mergeWorkflowExecutionOptions(emptyOptions, unpinnedOverrideOptions, asteriskUpdateMask)
-	assert.Error(t, err)
+	require.Error(t, err)
 }
 
 func TestMergeOptions_FooMask(t *testing.T) {
 	fooUpdateMask := &fieldmaskpb.FieldMask{Paths: []string{"foo"}}
 	_, err := mergeWorkflowExecutionOptions(emptyOptions, unpinnedOverrideOptions, fooUpdateMask)
-	assert.Error(t, err)
+	require.Error(t, err)
+}
+
+func TestMergeOptions_TimeSkippingConfig(t *testing.T) {
+	tscMask := &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}}
+	cfgA := &workflowpb.TimeSkippingConfig{Enabled: true}
+	cfgB := &workflowpb.TimeSkippingConfig{
+		Enabled: true,
+		Bound:   &workflowpb.TimeSkippingConfig_MaxSkippedDuration{MaxSkippedDuration: durationpb.New(time.Hour)},
+	}
+
+	tcs := []struct {
+		name        string
+		current     *workflowpb.WorkflowExecutionOptions
+		update      *workflowpb.WorkflowExecutionOptions
+		wantChanged bool
+		wantConfig  *workflowpb.TimeSkippingConfig
+	}{
+		// nil update means "don't touch" even when mask is present
+		{
+			name:        "nil update - existing config preserved",
+			current:     &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfgA},
+			update:      &workflowpb.WorkflowExecutionOptions{},
+			wantChanged: false,
+			wantConfig:  cfgA,
+		},
+		// non-nil update replaces and is detected as a change
+		{
+			name:        "new config - changed",
+			current:     &workflowpb.WorkflowExecutionOptions{},
+			update:      &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfgB},
+			wantChanged: true,
+			wantConfig:  cfgB,
+		},
+		// identical config is not detected as a change
+		{
+			name:        "same config - no change",
+			current:     &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfgB},
+			update:      &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfgB},
+			wantChanged: false,
+			wantConfig:  cfgB,
+		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			original := proto.Clone(tc.current).(*workflowpb.WorkflowExecutionOptions)
+			merged, err := mergeWorkflowExecutionOptions(tc.current, tc.update, tscMask)
+			require.NoError(t, err)
+			require.True(t, proto.Equal(tc.wantConfig, merged.GetTimeSkippingConfig()),
+				"config mismatch: want %v, got %v", tc.wantConfig, merged.GetTimeSkippingConfig())
+			require.Equal(t, tc.wantChanged, !proto.Equal(merged, original))
+		})
+	}
+}
+
+func TestMergeAndApply_TimeSkippingConfig(t *testing.T) {
+	tscMask := &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}}
+	cfg := &workflowpb.TimeSkippingConfig{
+		Enabled: true,
+		Bound:   &workflowpb.TimeSkippingConfig_MaxSkippedDuration{MaxSkippedDuration: durationpb.New(time.Hour)},
+	}
+
+	t.Run("same config - no event written", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := historyi.NewMockMutableState(ctrl)
+		ms.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
+			TimeSkippingInfo: &persistencespb.TimeSkippingInfo{Config: cfg},
+		}).AnyTimes()
+
+		merged, hasChanges, err := MergeAndApply(ms, &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfg}, tscMask, "")
+		require.NoError(t, err)
+		require.False(t, hasChanges)
+		require.True(t, proto.Equal(cfg, merged.GetTimeSkippingConfig()))
+	})
+
+	t.Run("new config - event written with config", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := historyi.NewMockMutableState(ctrl)
+		ms.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{}).AnyTimes()
+		ms.EXPECT().AddWorkflowExecutionOptionsUpdatedEvent(
+			nil, true, "", nil, nil, "", nil, cfg,
+		).Return(&historypb.HistoryEvent{}, nil)
+
+		merged, hasChanges, err := MergeAndApply(ms, &workflowpb.WorkflowExecutionOptions{TimeSkippingConfig: cfg}, tscMask, "")
+		require.NoError(t, err)
+		require.True(t, hasChanges)
+		require.True(t, proto.Equal(cfg, merged.GetTimeSkippingConfig()))
+	})
 }
 
 type (