Reapply Update on Reset

service/history/api/updateworkflow/api.go

@@ -26,6 +26,7 @@ package updateworkflow
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	commonpb "go.temporal.io/api/common/v1"
@@ -270,6 +271,7 @@ func (u *Updater) OnSuccess(
 	// If the long-poll times out due to serverTimeout then return a non-error empty response.
 	status, err := u.upd.WaitLifecycleStage(ctx, waitStage, serverTimeout)
 	if err != nil {
+		fmt.Println("update request wait error:", err)
 		return nil, err
 	}
 	resp := u.createResponse(u.wfKey, status.Outcome, status.Stage)

service/history/workflow/update/update.go

@@ -27,6 +27,7 @@ package update
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	enumspb "go.temporal.io/api/enums/v1"
@@ -632,6 +633,7 @@ func (u *Update) checkStateSet(msg proto.Message, allowed stateSet) error {
 
 // setState assigns the current state to a new value returning the original value.
 func (u *Update) setState(newState state) state {
+	fmt.Println("update state change", newState)
 	prevState := u.state
 	u.state = newState
 	u.instrumentation.stateChange(u.id, prevState, newState)

tests/reset_workflow.go

@@ -32,6 +32,7 @@
 	"strconv"
 	"time"
 
+	"github.com/davecgh/go-spew/spew"
 	"github.com/pborman/uuid"
 	commandpb "go.temporal.io/api/command/v1"
 	commonpb "go.temporal.io/api/common/v1"
@@ -760,6 +761,127 @@
 	}
 }
 
+func (s *FunctionalSuite) TestResetWorkflow_Update_ReapplyBufferAll() {
+	tv := testvars.New(s.T())
+	s.testResetWorkflowUpdateReapplyBuffer(tv, enumspb.RESET_REAPPLY_TYPE_ALL_ELIGIBLE)
+}
+
+func (s *FunctionalSuite) TestResetWorkflow_Update_ReapplyBufferNone() {
+	tv := testvars.New(s.T())
+	s.testResetWorkflowUpdateReapplyBuffer(tv, enumspb.RESET_REAPPLY_TYPE_NONE)
+}
+
+func (s *FunctionalSuite) testResetWorkflowUpdateReapplyBuffer(
+	tv *testvars.TestVars,
+	reapplyType enumspb.ResetReapplyType,
+) {
+	tv = s.startWorkflow(tv)
+	s.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(tv.RunID()))
+
+	var resetRunID string
+	var updResult <-chan updateResponseErr
+	wtHandlerCalls := 0
+	wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
+		spew.Dump(task.History.Events)
+
+		wtHandlerCalls++
+		switch wtHandlerCalls {
+		case 1:
+			s.EqualHistoryEvents(`
+  1 WorkflowExecutionStarted
+  2 WorkflowTaskScheduled
+  3 WorkflowTaskStarted`, task.History.Events)
+
+			// (1) send Update
+			updResult = s.sendUpdate(NewContext(), tv, "1")
+
+			// (2) send Reset
+			resp, err := s.client.ResetWorkflowExecution(NewContext(),
+				&workflowservice.ResetWorkflowExecutionRequest{
+					Namespace: s.namespace,
+					WorkflowExecution: &commonpb.WorkflowExecution{
+						WorkflowId: tv.WorkflowID(),
+						RunId:      tv.RunID(),
+					},
+					Reason:                    "reset execution from test",
+					WorkflowTaskFinishEventId: 3,
+					RequestId:                 uuid.New(),
+					ResetReapplyType:          reapplyType,
+				})
+			s.NoError(err)
+			resetRunID = resp.RunId
+		case 2:
+			//return s.UpdateAcceptCompleteCommands(tv, "1"), nil
+		}
+
+		return []*commandpb.Command{{
+			CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
+			Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
+				CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
+					Result: payloads.EncodeString("Done"),
+				},
+			},
+		}}, nil
+	}
+
+	msgHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*protocolpb.Message, error) {
+		if len(task.Messages) > 0 {
+			return s.UpdateAcceptCompleteMessages(tv, task.Messages[0], "1"), nil
+		}
+		return nil, nil
+	}
+
+	poller := &TaskPoller{
+		Client:              s.client,
+		Namespace:           s.namespace,
+		TaskQueue:           tv.TaskQueue(),
+		Identity:            tv.WorkerIdentity(),
+		WorkflowTaskHandler: wtHandler,
+		MessageHandler:      msgHandler,
+		Logger:              s.Logger,
+		T:                   s.T(),
+	}
+
+	_, err := poller.PollAndProcessWorkflowTask()
+	s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
+	s.Error(err) // due to workflow termination (reset)
+
+	_, err = poller.PollAndProcessWorkflowTask()
+	s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
+	s.NoError(err)
+
+	spew.Dump("update result", <-updResult)
+
+	events := s.getHistory(s.namespace, &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: resetRunID})
+	switch reapplyType {
+	case enumspb.RESET_REAPPLY_TYPE_ALL_ELIGIBLE:
+		s.EqualHistoryEvents(`
+  1 WorkflowExecutionStarted
+  2 WorkflowTaskScheduled
+  3 WorkflowTaskStarted
+  4 WorkflowTaskFailed
+  5 WorkflowExecutionSignaled  // reapplied Update
+  6 WorkflowTaskScheduled
+  7 WorkflowTaskStarted
+  8 WorkflowTaskCompleted
+  9 WorkflowExecutionCompleted
+`, events)
+	case enumspb.RESET_REAPPLY_TYPE_NONE:
+		s.EqualHistoryEvents(`
+  1 WorkflowExecutionStarted
+  2 WorkflowTaskScheduled
+  3 WorkflowTaskStarted
+  4 WorkflowTaskFailed
+  5 WorkflowTaskScheduled  // no reapplied Update
+  6 WorkflowTaskStarted
+  7 WorkflowTaskCompleted
+  8 WorkflowExecutionCompleted
+`, events)
+	default:
+		panic(fmt.Sprintf("unknown reset reapply type: %v", reapplyType))
+	}
+}
+
 func (s *FunctionalSuite) TestResetWorkflow_WorkflowTask_Schedule() {
 	workflowID := "functional-reset-workflow-test-schedule"
 	workflowTypeName := "functional-reset-workflow-test-schedule-type"