Wait for local activity while holding workflow context lock (#760)

Ensure query decisions are serialized with other decision processing including local activities.
* move wait for local activity into process workflow task - holding the context lock
* implement decision heartbeat logic while holding lock

Author: shreyassrivatsan

internal/internal_public.go

@@ -34,6 +34,8 @@ import (
 )
 
 type (
+	decisionHeartbeatFunc func(response interface{}, startTime time.Time) (*workflowTask, error)
+
 	// HistoryIterator iterator through history events
 	HistoryIterator interface {
 		// GetNextPage returns next page of history events
@@ -74,7 +76,10 @@ type (
 		// - RespondDecisionTaskCompletedRequest
 		// - RespondDecisionTaskFailedRequest
 		// - RespondQueryTaskCompletedRequest
-		ProcessWorkflowTask(task *workflowTask) (response interface{}, w WorkflowExecutionContext, err error)
+		ProcessWorkflowTask(
+			task *workflowTask,
+			f decisionHeartbeatFunc,
+		) (response interface{}, err error)
 	}
 
 	// ActivityTaskHandler represents activity task handlers.

internal/internal_task_handlers.go

@@ -148,6 +148,10 @@ type (
 		currentIndex  int
 		next          []*s.HistoryEvent
 	}
+
+	decisionHeartbeatError struct {
+		Message string
+	}
 )
 
 func newHistory(task *workflowTask, eventsHandler *workflowExecutionEventHandlerImpl) *history {
@@ -161,6 +165,10 @@ func newHistory(task *workflowTask, eventsHandler *workflowExecutionEventHandler
 	return result
 }
 
+func (e decisionHeartbeatError) Error() string {
+	return e.Message
+}
+
 // Get workflow start event.
 func (eh *history) GetWorkflowStartedEvent() (*s.HistoryEvent, error) {
 	events := eh.workflowTask.task.History.Events
@@ -545,8 +553,10 @@ func (wth *workflowTaskHandlerImpl) createWorkflowContext(task *s.PollForDecisio
 	return workflowContext, nil
 }
 
-func (wth *workflowTaskHandlerImpl) getOrCreateWorkflowContext(task *s.PollForDecisionTaskResponse,
-	historyIterator HistoryIterator) (workflowContext *workflowExecutionContextImpl, err error) {
+func (wth *workflowTaskHandlerImpl) getOrCreateWorkflowContext(
+	task *s.PollForDecisionTaskResponse,
+	historyIterator HistoryIterator,
+) (workflowContext *workflowExecutionContextImpl, err error) {
 	metricsScope := wth.metricsScope.GetTaggedScope(tagWorkflowType, task.WorkflowType.GetName())
 	defer func() {
 		if err == nil && workflowContext != nil && workflowContext.laTunnel == nil {
@@ -631,9 +641,10 @@ func (w *workflowExecutionContextImpl) resetStateIfDestroyed(task *s.PollForDeci
 // ProcessWorkflowTask processes all the events of the workflow task.
 func (wth *workflowTaskHandlerImpl) ProcessWorkflowTask(
 	workflowTask *workflowTask,
-) (completeRequest interface{}, context WorkflowExecutionContext, errRet error) {
+	heartbeatFunc decisionHeartbeatFunc,
+) (completeRequest interface{}, errRet error) {
 	if workflowTask == nil || workflowTask.task == nil {
-		return nil, nil, errors.New("nil workflow task provided")
+		return nil, errors.New("nil workflow task provided")
 	}
 	task := workflowTask.task
 	if task.History == nil || len(task.History.Events) == 0 {
@@ -642,7 +653,7 @@ func (wth *workflowTaskHandlerImpl) ProcessWorkflowTask(
 		}
 	}
 	if task.Query == nil && len(task.History.Events) == 0 {
-		return nil, nil, errors.New("nil or empty history")
+		return nil, errors.New("nil or empty history")
 	}
 
 	runID := task.WorkflowExecution.GetRunId()
@@ -657,15 +668,53 @@ func (wth *workflowTaskHandlerImpl) ProcessWorkflowTask(
 
 	workflowContext, err := wth.getOrCreateWorkflowContext(task, workflowTask.historyIterator)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	defer func() {
 		workflowContext.Unlock(errRet)
 	}()
 
-	response, err := workflowContext.ProcessWorkflowTask(workflowTask)
-	return response, workflowContext, err
+	var response interface{}
+process_Workflow_Loop:
+	for {
+		startTime := time.Now()
+		response, err = workflowContext.ProcessWorkflowTask(workflowTask)
+		if err == nil && response == nil {
+		wait_LocalActivity_Loop:
+			for {
+				deadlineToTrigger := time.Duration(float32(ratioToForceCompleteDecisionTaskComplete) * float32(workflowContext.GetDecisionTimeout()))
+				delayDuration := startTime.Add(deadlineToTrigger).Sub(time.Now())
+				select {
+				case <-time.After(delayDuration):
+					// force complete, call the decision heartbeat function
+					workflowTask, err = heartbeatFunc(
+						workflowContext.CompleteDecisionTask(workflowTask, false),
+						startTime,
+					)
+					if err != nil {
+						return nil, &decisionHeartbeatError{Message: fmt.Sprintf("error sending decision heartbeat %v", err)}
+					}
+					if workflowTask == nil {
+						return nil, nil
+					}
+					continue process_Workflow_Loop
+
+				case lar := <-workflowTask.laResultCh:
+					// local activity result ready
+					response, err = workflowContext.ProcessLocalActivityResult(workflowTask, lar)
+					if err == nil && response == nil {
+						// decision task is not done yet, still waiting for more local activities
+						continue wait_LocalActivity_Loop
+					}
+					break process_Workflow_Loop
+				}
+			}
+		} else {
+			break process_Workflow_Loop
+		}
+	}
+	return response, err
 }
 
 func (w *workflowExecutionContextImpl) ProcessWorkflowTask(workflowTask *workflowTask) (interface{}, error) {
@@ -956,10 +1005,6 @@ func (w *workflowExecutionContextImpl) skipReplayCheck() bool {
 	return w.currentDecisionTask.Query != nil || !isFullHistory(w.currentDecisionTask.History)
 }
 
-func (w *workflowExecutionContextImpl) GetCurrentDecisionTask() *s.PollForDecisionTaskResponse {
-	return w.currentDecisionTask
-}
-
 func (w *workflowExecutionContextImpl) SetCurrentTask(task *s.PollForDecisionTaskResponse) {
 	w.currentDecisionTask = task
 	// do not update the previousStartedEventID for query task
@@ -993,13 +1038,6 @@ func (w *workflowExecutionContextImpl) GetDecisionTimeout() time.Duration {
 	return time.Second * time.Duration(w.workflowInfo.TaskStartToCloseTimeoutSeconds)
 }
 
-func (w *workflowExecutionContextImpl) StackTrace() string {
-	if w.eventHandler == nil {
-		return "eventHandler is closed"
-	}
-	return w.eventHandler.StackTrace()
-}
-
 func skipDeterministicCheckForDecision(d *s.Decision) bool {
 	if d.GetDecisionType() == s.DecisionTypeRecordMarker {
 		markerName := d.RecordMarkerDecisionAttributes.GetMarkerName()

internal/internal_task_handlers_interfaces_test.go

@@ -21,14 +21,15 @@
 package internal
 
 import (
+	"testing"
+
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/suite"
 	"github.com/uber/tchannel-go/thrift"
 	"go.uber.org/cadence/.gen/go/cadence/workflowservicetest"
 	m "go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/internal/common"
 	"golang.org/x/net/context"
-	"testing"
 )
 
 type (
@@ -44,10 +45,12 @@ type sampleWorkflowTaskHandler struct {
 }
 
 func (wth sampleWorkflowTaskHandler) ProcessWorkflowTask(
-	workflowTask *workflowTask) (interface{}, WorkflowExecutionContext, error) {
+	workflowTask *workflowTask,
+	d decisionHeartbeatFunc,
+) (interface{}, error) {
 	return &m.RespondDecisionTaskCompletedRequest{
 		TaskToken: workflowTask.task.TaskToken,
-	}, nil, nil
+	}, nil
 }
 
 func newSampleWorkflowTaskHandler() *sampleWorkflowTaskHandler {
@@ -104,7 +107,7 @@ func (s *PollLayerInterfacesTestSuite) TestProcessWorkflowTaskInterface() {
 
 	// Process task and respond to the service.
 	taskHandler := newSampleWorkflowTaskHandler()
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: response})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: response}, nil)
 	completionRequest := request.(*m.RespondDecisionTaskCompletedRequest)
 	s.NoError(err)
 

internal/internal_task_handlers_test.go

@@ -234,7 +234,7 @@ func (t *TaskHandlersTestSuite) testWorkflowTaskWorkflowExecutionStartedHelper(p
 	}
 	task := createWorkflowTask(testEvents, 0, "HelloWorld_Workflow")
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)
@@ -286,7 +286,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_ActivityTaskScheduled() {
 		Logger:   t.logger,
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	response := request.(*s.RespondDecisionTaskCompletedRequest)
 
 	t.NoError(err)
@@ -297,7 +297,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_ActivityTaskScheduled() {
 
 	// Schedule an activity and see if we complete workflow, Having only one last decision.
 	task = createWorkflowTask(testEvents, 3, "HelloWorld_Workflow")
-	request, _, err = taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err = taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	response = request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)
@@ -337,7 +337,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_QueryWorkflow_Sticky() {
 	task := createWorkflowTask(testEvents[0:1], 0, "HelloWorld_Workflow")
 	task.StartedEventId = common.Int64Ptr(1)
 	task.WorkflowExecution = execution
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)
@@ -348,7 +348,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_QueryWorkflow_Sticky() {
 	// then check the current state using query task
 	task = createQueryTask([]*s.HistoryEvent{}, 6, "HelloWorld_Workflow", "test-query")
 	task.WorkflowExecution = execution
-	queryResp, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	queryResp, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.NoError(err)
 	t.verifyQueryResult(queryResp, "waiting-activity-result")
 }
@@ -380,30 +380,30 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_QueryWorkflow_NonSticky() {
 	// query after first decision task (notice the previousStartEventID is always the last eventID for query task)
 	task := createQueryTask(testEvents[0:3], 3, "HelloWorld_Workflow", "test-query")
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, _ := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	response, _ := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.verifyQueryResult(response, "waiting-activity-result")
 
 	// query after activity task complete but before second decision task started
 	task = createQueryTask(testEvents[0:7], 7, "HelloWorld_Workflow", "test-query")
 	taskHandler = newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, _ = taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	response, _ = taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.verifyQueryResult(response, "waiting-activity-result")
 
 	// query after second decision task
 	task = createQueryTask(testEvents[0:8], 8, "HelloWorld_Workflow", "test-query")
 	taskHandler = newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, _ = taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	response, _ = taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.verifyQueryResult(response, "done")
 
 	// query after second decision task with extra events
 	task = createQueryTask(testEvents[0:9], 9, "HelloWorld_Workflow", "test-query")
 	taskHandler = newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, _ = taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	response, _ = taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.verifyQueryResult(response, "done")
 
 	task = createQueryTask(testEvents[0:9], 9, "HelloWorld_Workflow", "invalid-query-type")
 	taskHandler = newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, _ = taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	response, _ = taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.NotNil(response)
 	queryResp, ok := response.(*s.RespondQueryTaskCompletedRequest)
 	t.True(ok)
@@ -451,7 +451,7 @@ func (t *TaskHandlersTestSuite) TestCacheEvictionWhenErrorOccurs() {
 	// newWorkflowTaskWorkerInternal will set the laTunnel in taskHandler, without it, ProcessWorkflowTask()
 	// will fail as it can't find laTunnel in getWorkflowCache().
 	newWorkflowTaskWorkerInternal(taskHandler, t.service, testDomain, params, make(chan struct{}))
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 
 	t.Error(err)
 	t.Nil(request)
@@ -507,7 +507,7 @@ func (t *TaskHandlersTestSuite) testSideEffectDeferHelper(disableSticky bool) {
 
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
 	task := createWorkflowTask(testEvents, 0, workflowName)
-	_, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	_, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.Nil(err)
 
 	if !params.DisableStickyExecution {
@@ -549,7 +549,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicDetection() {
 	}
 
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	// there should be no error as the history events matched the decisions.
 	t.NoError(err)
@@ -561,7 +561,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicDetection() {
 	// newWorkflowTaskWorkerInternal will set the laTunnel in taskHandler, without it, ProcessWorkflowTask()
 	// will fail as it can't find laTunnel in getWorkflowCache().
 	newWorkflowTaskWorkerInternal(taskHandler, t.service, testDomain, params, stopC)
-	request, _, err = taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err = taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.Error(err)
 	t.Nil(request)
 	t.Contains(err.Error(), "nondeterministic")
@@ -571,7 +571,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicDetection() {
 	params.NonDeterministicWorkflowPolicy = NonDeterministicWorkflowPolicyFailWorkflow
 	failOnNondeterminismTaskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
 	task = createWorkflowTask(testEvents, 3, "HelloWorld_Workflow")
-	request, _, err = failOnNondeterminismTaskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err = failOnNondeterminismTaskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	// When FailWorkflow policy is set, task handler does not return an error,
 	// because it will indicate non determinism in the request.
 	t.NoError(err)
@@ -589,7 +589,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicDetection() {
 	// now with different package name to activity type
 	testEvents[4].ActivityTaskScheduledEventAttributes.ActivityType.Name = common.StringPtr("new-package.Greeter_Activity")
 	task = createWorkflowTask(testEvents, 3, "HelloWorld_Workflow")
-	request, _, err = taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err = taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.NoError(err)
 	t.NotNil(request)
 }
@@ -610,7 +610,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_WorkflowReturnsPanicError() {
 	}
 
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.NoError(err)
 	t.NotNil(request)
 	r, ok := request.(*s.RespondDecisionTaskCompletedRequest)
@@ -638,7 +638,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_WorkflowPanics() {
 	}
 
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.NoError(err)
 	t.NotNil(request)
 	r, ok := request.(*s.RespondDecisionTaskFailedRequest)
@@ -690,7 +690,7 @@ func (t *TaskHandlersTestSuite) TestGetWorkflowInfo() {
 	}
 
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	t.NoError(err)
 	t.NotNil(request)
 	r, ok := request.(*s.RespondDecisionTaskCompletedRequest)
@@ -728,11 +728,10 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_CancelActivityBeforeSent() {
 		Logger:   t.logger,
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task}, nil)
 	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)
-	//t.printAllDecisions(response.Decisions)
 	t.Equal(1, len(response.Decisions))
 	t.Equal(s.DecisionTypeCompleteWorkflowExecution, response.Decisions[0].GetDecisionType())
 	t.NotNil(response.Decisions[0].CompleteWorkflowExecutionDecisionAttributes)
@@ -764,7 +763,7 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_PageToken() {
 		},
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	request, _, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task, historyIterator: historyIterator})
+	request, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task, historyIterator: historyIterator}, nil)
 	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)