Implement post reset operations (#7719)

## What changed?
This PR adds PostResetOperation support to reset API. With this we will
be performing certain operations after reset. In this PR
  - Added support for performing workflow options update. 
  - Updated WorkflowResetter interface.
  - Added functional test 

## Why?
Developers/Operators often want to perform certain operations soon after
resetting a workflow. Ex: updating worker version, sending signal to
workflow etc. These operations need to be performed atomically along
with the reset operation.

## 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)

Author: gow

service/history/api/reapplyevents/api.go

@@ -138,6 +138,7 @@ func Invoke(
 					toReapplyEvents,
 					nil,
 					false, // allowResetWithPendingChildren
+					nil,
 				)
 				switch err.(type) {
 				case *serviceerror.InvalidArgument:

service/history/api/resetworkflow/api.go

@@ -2,17 +2,20 @@ package resetworkflow
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/google/uuid"
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
+	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/definition"
 	"go.temporal.io/server/common/locks"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence/versionhistory"
+	"go.temporal.io/server/common/worker_versioning"
 	"go.temporal.io/server/service/history/api"
 	historyi "go.temporal.io/server/service/history/interfaces"
 	"go.temporal.io/server/service/history/ndc"
@@ -30,6 +33,10 @@ func Invoke(
 		return nil, err
 	}
 
+	if err := validatePostResetOperationInputs(resetRequest.ResetRequest.PostResetOperations); err != nil {
+		return nil, err
+	}
+
 	request := resetRequest.ResetRequest
 	workflowID := request.WorkflowExecution.GetWorkflowId()
 	baseRunID := request.WorkflowExecution.GetRunId()
@@ -151,6 +158,7 @@ func Invoke(
 		nil,
 		GetResetReapplyExcludeTypes(request.GetResetReapplyExcludeTypes(), request.GetResetReapplyType()),
 		allowResetWithPendingChildren,
+		resetRequest.ResetRequest.PostResetOperations,
 	); err != nil {
 		return nil, err
 	}
@@ -186,3 +194,19 @@ func GetResetReapplyExcludeTypes(
 	}
 	return exclude
 }
+
+// validatePostResetOperationInputs validates the optional post reset operation inputs.
+func validatePostResetOperationInputs(postResetOperations []*workflowpb.PostResetOperation) error {
+	for _, operation := range postResetOperations {
+		switch op := operation.GetVariant().(type) {
+		case *workflowpb.PostResetOperation_UpdateWorkflowOptions_:
+			opts := op.UpdateWorkflowOptions.GetWorkflowExecutionOptions()
+			if err := worker_versioning.ValidateVersioningOverride(opts.GetVersioningOverride()); err != nil {
+				return err
+			}
+		default:
+			return serviceerror.NewInvalidArgument(fmt.Sprintf("unsupported post reset operation: %T", op))
+		}
+	}
+	return nil
+}

service/history/history_engine_test.go

@@ -5291,7 +5291,7 @@ func (s *engineSuite) TestReapplyEvents_ResetWorkflow() {
 		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
-		gomock.Any(),
+		gomock.Any(), gomock.Any(),
 	).Return(nil)
 
 	err = s.historyEngine.ReapplyEvents(

service/history/ndc/transaction_manager.go

@@ -326,6 +326,7 @@ func (r *transactionMgrImpl) backfillWorkflowEventsReapply(
 			totalEvents,
 			nil,
 			false, // allowResetWithPendingChildren
+			nil,
 		)
 		switch err.(type) {
 		case *serviceerror.InvalidArgument:

service/history/ndc/transaction_manager_test.go

@@ -226,6 +226,8 @@ func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Active_Closed
 		workflowEvents.Events,
 		nil,
 		false, // allowResetWithPendingChildren
+		nil,   // post reset operations
+
 	).Return(nil)
 
 	s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{
@@ -307,6 +309,7 @@ func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Closed_ResetF
 		workflowEvents.Events,
 		nil,
 		false, // allowResetWithPendingChildren
+		nil,   // post reset operations
 	).Return(serviceerror.NewInvalidArgument("reset fail"))
 
 	s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{

service/history/ndc/workflow_resetter.go

@@ -12,6 +12,7 @@ import (
 	enumspb "go.temporal.io/api/enums/v1"
 	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
+	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
@@ -26,6 +27,7 @@ import (
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/util"
+	"go.temporal.io/server/service/history/api/updateworkflowoptions"
 	"go.temporal.io/server/service/history/consts"
 	"go.temporal.io/server/service/history/hsm"
 	historyi "go.temporal.io/server/service/history/interfaces"
@@ -64,6 +66,7 @@ type (
 			additionalReapplyEvents []*historypb.HistoryEvent,
 			resetReapplyExcludeTypes map[enumspb.ResetReapplyExcludeType]struct{},
 			allowResetWithPendingChildren bool,
+			postResetOperations []*workflowpb.PostResetOperation,
 		) error
 	}
 
@@ -119,6 +122,7 @@ func (r *workflowResetterImpl) ResetWorkflow(
 	additionalReapplyEvents []*historypb.HistoryEvent,
 	resetReapplyExcludeTypes map[enumspb.ResetReapplyExcludeType]struct{},
 	allowResetWithPendingChildren bool,
+	postResetOperations []*workflowpb.PostResetOperation,
 ) (retError error) {
 
 	namespaceEntry, err := r.namespaceRegistry.GetNamespaceByID(namespaceID)
@@ -225,6 +229,10 @@ func (r *workflowResetterImpl) ResetWorkflow(
 		return err
 	}
 
+	if err := r.performPostResetOperations(ctx, resetMS, postResetOperations); err != nil {
+		return err
+	}
+
 	if err := workflow.ScheduleWorkflowTask(resetMS); err != nil {
 		return err
 	}
@@ -1136,3 +1144,17 @@ func (r *workflowResetterImpl) shouldExcludeAllReapplyEvents(excludeTypes map[en
 	}
 	return true
 }
+
+// performPostResetOperations performs the optional post reset operations on the reset workflow.
+func (r *workflowResetterImpl) performPostResetOperations(ctx context.Context, resetMS historyi.MutableState, postResetOperations []*workflowpb.PostResetOperation) error {
+	for _, operation := range postResetOperations {
+		switch op := operation.GetVariant().(type) {
+		case *workflowpb.PostResetOperation_UpdateWorkflowOptions_:
+			_, _, err := updateworkflowoptions.MergeAndApply(resetMS, op.UpdateWorkflowOptions.GetWorkflowExecutionOptions(), op.UpdateWorkflowOptions.GetUpdateMask())
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}

service/history/ndc/workflow_resetter_mock.go

@@ -1615,6 +1615,7 @@ func (t *transferQueueActiveTaskExecutor) resetWorkflow(
 		nil,
 		nil,
 		allowResetWithPendingChildren,
+		nil,
 	)
 
 	switch err.(type) {

service/history/transfer_queue_active_task_executor.go

@@ -10,13 +10,16 @@ import (
 	commandpb "go.temporal.io/api/command/v1"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
 	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/client"
 	"go.temporal.io/server/api/adminservice/v1"
 	"go.temporal.io/server/common/payloads"
 	"go.temporal.io/server/tests/testcore"
 	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/fieldmaskpb"
 )
 
 // Tests workflow reset feature
@@ -193,6 +196,59 @@ func (s *WorkflowResetSuite) TestOriginalExecutionRunId() {
 	}
 }
 
+// Test that the workflow options are updated when the workflow is reset.
+func (s *WorkflowResetSuite) TestResetWorkflowWithOptionsUpdate() {
+	workflowID := "test-reset" + uuid.NewString()
+	ctx := testcore.NewContext()
+	runs := s.setupRuns(ctx, workflowID, 1, true)
+	currentRunID := runs[0]
+
+	// Reset the workflow by providing the explicit runID (base run) to reset.
+	resp, err := s.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
+		Namespace:                 s.Namespace().String(),
+		WorkflowExecution:         &commonpb.WorkflowExecution{WorkflowId: workflowID, RunId: currentRunID},
+		Reason:                    "testing-reset",
+		RequestId:                 uuid.NewString(),
+		WorkflowTaskFinishEventId: s.getFirstWFTaskCompleteEventID(ctx, workflowID, currentRunID),
+		PostResetOperations: []*workflowpb.PostResetOperation{
+			{
+				Variant: &workflowpb.PostResetOperation_UpdateWorkflowOptions_{
+					UpdateWorkflowOptions: &workflowpb.PostResetOperation_UpdateWorkflowOptions{
+						WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{
+							VersioningOverride: &workflowpb.VersioningOverride{
+								Behavior:      enumspb.VERSIONING_BEHAVIOR_PINNED,
+								PinnedVersion: "testing.v.123",
+							},
+						},
+						UpdateMask: &fieldmaskpb.FieldMask{
+							Paths: []string{
+								"versioning_override",
+							},
+						},
+					},
+				},
+			},
+		},
+	})
+	s.NoError(err)
+	newRunID := resp.RunId
+
+	// assert that the new run has the updated workflow options
+	var optionsUpdatedEvent *historypb.HistoryEvent
+	hist := s.SdkClient().GetWorkflowHistory(ctx, workflowID, newRunID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
+	for hist.HasNext() {
+		event, err := hist.Next()
+		s.NoError(err)
+		if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED {
+			optionsUpdatedEvent = event
+			break
+		}
+	}
+	s.NotNil(optionsUpdatedEvent)
+	s.Equal(optionsUpdatedEvent.GetWorkflowExecutionOptionsUpdatedEventAttributes().GetVersioningOverride().GetBehavior(), enumspb.VERSIONING_BEHAVIOR_PINNED)
+	s.Equal(optionsUpdatedEvent.GetWorkflowExecutionOptionsUpdatedEventAttributes().GetVersioningOverride().GetPinnedVersion(), "testing.v.123")
+}
+
 // Helper methods
 
 // getFirstWFTaskCompleteEventID finds the first event corresponding to workflow task completion. This can be used as a good reset point for tests in this suite.