query API client impl (#226)

* query API client impl

* update based on comments

* add unit test to cover case where history contains more events after decision task started

Author: yiminc

client.go

@@ -29,6 +29,10 @@ import (
 	"go.uber.org/cadence/common/metrics"
 )
 
+// QueryTypeStackTrace is the build in query type for Client.QueryWorkflow() call. Use this query type to get the call
+// stack of the workflow. The result will be a string encoded in the EncodedValue.
+const QueryTypeStackTrace string = "__stack_trace"
+
 type (
 	// Client is the client for starting and getting information about a workflow executions as well as
 	// completing activities asynchronously.
@@ -127,6 +131,26 @@ type (
 		//  - InternalServiceError
 		//  - EntityNotExistError
 		ListOpenWorkflow(request *s.ListOpenWorkflowExecutionsRequest) (*s.ListOpenWorkflowExecutionsResponse, error)
+
+		// QueryWorkflow queries a given workflow execution and returns the query result synchronously. Parameter workflowID
+		// and queryType are required, other parameters are optional. The workflowID and runID (optional) identify the
+		// target workflow execution that this query will be send to. If runID is not specified (empty string), server will
+		// use the currently running execution of that workflowID. The queryType specifies the type of query you want to
+		// run. By default, cadence supports "__stack_trace" as a standard query type, which will return string value
+		// representing the call stack of the target workflow. The target workflow could also setup different query handler
+		// to handle custom query types.
+		// See comments at cadence.SetQueryHandler(ctx Context, queryType string, handler interface{}) for more details
+		// on how to setup query handler within the target workflow.
+		// - workflowID is required.
+		// - runID can be default(empty string). if empty string then it will pick the running execution of that workflow ID.
+		// - queryType is the type of the query.
+		// - args... are the optional query parameters.
+		// The errors it can return:
+		//  - BadRequestError
+		//  - InternalServiceError
+		//  - EntityNotExistError
+		//  - QueryFailError
+		QueryWorkflow(workflowID string, runID string, queryType string, args ...interface{}) (EncodedValue, error)
 	}
 
 	// ClientOptions are optional parameters for Client creation.

internal_event_handlers.go

@@ -82,6 +82,7 @@ type (
 		completeHandler completionHandler               // events completion handler
 		cancelHandler   func()                          // A cancel handler to be invoked on a cancel notification
 		signalHandler   func(name string, input []byte) // A signal handler to be invoked on a signal event
+		queryHandler    func(queryType string, queryArgs []byte) ([]byte, error)
 
 		logger                *zap.Logger
 		isReplay              bool // flag to indicate if workflow is in replay mode
@@ -241,6 +242,10 @@ func (wc *workflowEnvironmentImpl) RegisterSignalHandler(handler func(name strin
 	wc.signalHandler = handler
 }
 
+func (wc *workflowEnvironmentImpl) RegisterQueryHandler(handler func(string, []byte) ([]byte, error)) {
+	wc.queryHandler = handler
+}
+
 func (wc *workflowEnvironmentImpl) GetLogger() *zap.Logger {
 	return wc.logger
 }
@@ -572,6 +577,13 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 	return weh.decisionsHelper.getDecisions(true), nil
 }
 
+func (weh *workflowExecutionEventHandlerImpl) ProcessQuery(queryType string, queryArgs []byte) ([]byte, error) {
+	if queryType == QueryTypeStackTrace {
+		return getHostEnvironment().encodeArg(weh.StackTrace())
+	}
+	return weh.queryHandler(queryType, queryArgs)
+}
+
 func (weh *workflowExecutionEventHandlerImpl) StackTrace() string {
 	return weh.workflowDefinition.StackTrace()
 }

internal_public.go

@@ -45,7 +45,7 @@ type (
 		ProcessWorkflowTask(
 			task *s.PollForDecisionTaskResponse,
 			getHistoryPage GetHistoryPage,
-			emitStack bool) (response *s.RespondDecisionTaskCompletedRequest, stackTrace string, err error)
+			emitStack bool) (response interface{}, stackTrace string, err error)
 	}
 
 	// ActivityTaskHandler represents activity task handlers.

internal_task_handlers.go

@@ -52,6 +52,8 @@ type (
 		// Process a single event and return the assosciated decisions.
 		// Return List of decisions made, any error.
 		ProcessEvent(event *s.HistoryEvent, isReplay bool, isLast bool) ([]*s.Decision, error)
+		// ProcessQuery process a query request.
+		ProcessQuery(queryType string, queryArgs []byte) ([]byte, error)
 		StackTrace() string
 		// Close for cleaning up resources on this event handler
 		Close()
@@ -319,7 +321,7 @@ func (wth *workflowTaskHandlerImpl) ProcessWorkflowTask(
 	task *s.PollForDecisionTaskResponse,
 	getHistoryPage GetHistoryPage,
 	emitStack bool,
-) (result *s.RespondDecisionTaskCompletedRequest, stackTrace string, err error) {
+) (result interface{}, stackTrace string, err error) {
 	if task == nil {
 		return nil, "", errors.New("nil workflowtask provided")
 	}
@@ -439,10 +441,13 @@ ProcessEvents:
 			}
 		}
 	}
-	// check if decisions from reply matches to the history events
-	if err := matchReplayWithHistory(replayDecisions, respondEvents); err != nil {
-		wth.logger.Error("Replay and history mismatch.", zap.Error(err))
-		return nil, "", err
+
+	if task.Query == nil {
+		// check if decisions from reply matches to the history events
+		if err := matchReplayWithHistory(replayDecisions, respondEvents); err != nil {
+			wth.logger.Error("Replay and history mismatch.", zap.Error(err))
+			return nil, "", err
+		}
 	}
 
 	startEvent, err := reorderedHistory.GetWorkflowStartedEvent()
@@ -482,28 +487,44 @@ ProcessEvents:
 		}
 	}
 
-	// Fill the response.
-	taskCompletionRequest := &s.RespondDecisionTaskCompletedRequest{
-		TaskToken: task.TaskToken,
-		Decisions: decisions,
-		Identity:  common.StringPtr(wth.identity),
-		// ExecutionContext:
-	}
-
-	traceLog(func() {
-		var buf bytes.Buffer
-		for i, d := range decisions {
-			buf.WriteString(fmt.Sprintf("%v: %v\n", i, util.DecisionToString(d)))
+	var completeRequest interface{}
+	if task.Query != nil {
+		// for query task
+		result, err := eventHandler.ProcessQuery(task.Query.GetQueryType(), task.Query.GetQueryArgs_())
+		queryTaskCompleteRequest := &s.RespondQueryTaskCompletedRequest{
+			TaskToken: task.TaskToken,
 		}
-		wth.logger.Debug("new_decisions",
-			zap.Int("DecisionCount", len(decisions)),
-			zap.String("Decisions", buf.String()))
-	})
+		if err != nil {
+			queryTaskCompleteRequest.CompletedType = s.QueryTaskCompletedTypePtr(s.QueryTaskCompletedType_FAILED)
+			queryTaskCompleteRequest.ErrorMessage = common.StringPtr(err.Error())
+		} else {
+			queryTaskCompleteRequest.CompletedType = s.QueryTaskCompletedTypePtr(s.QueryTaskCompletedType_COMPLETED)
+			queryTaskCompleteRequest.QueryResult_ = result
+		}
+		completeRequest = queryTaskCompleteRequest
+	} else {
+		// Fill the response.
+		completeRequest = &s.RespondDecisionTaskCompletedRequest{
+			TaskToken: task.TaskToken,
+			Decisions: decisions,
+			Identity:  common.StringPtr(wth.identity),
+			// ExecutionContext:
+		}
+		traceLog(func() {
+			var buf bytes.Buffer
+			for i, d := range decisions {
+				buf.WriteString(fmt.Sprintf("%v: %v\n", i, util.DecisionToString(d)))
+			}
+			wth.logger.Debug("new_decisions",
+				zap.Int("DecisionCount", len(decisions)),
+				zap.String("Decisions", buf.String()))
+		})
+	}
 
 	if emitStack {
 		stackTrace = eventHandler.StackTrace()
 	}
-	return taskCompletionRequest, stackTrace, nil
+	return completeRequest, stackTrace, nil
 }
 
 func isVersionMarkerDecision(d *s.Decision) bool {

internal_task_handlers_interfaces_test.go

@@ -70,7 +70,7 @@ type sampleWorkflowTaskHandler struct {
 func (wth sampleWorkflowTaskHandler) ProcessWorkflowTask(
 	task *m.PollForDecisionTaskResponse,
 	getHistoryPage GetHistoryPage,
-	emitStack bool) (*m.RespondDecisionTaskCompletedRequest, string, error) {
+	emitStack bool) (interface{}, string, error) {
 	return &m.RespondDecisionTaskCompletedRequest{
 		TaskToken: task.TaskToken,
 	}, "", nil
@@ -126,7 +126,8 @@ func (s *PollLayerInterfacesTestSuite) TestProcessWorkflowTaskInterface() {
 
 	// Process task and respond to the service.
 	taskHandler := newSampleWorkflowTaskHandler()
-	completionRequest, _, err := taskHandler.ProcessWorkflowTask(response, nil, false)
+	request, _, err := taskHandler.ProcessWorkflowTask(response, nil, false)
+	completionRequest := request.(*m.RespondDecisionTaskCompletedRequest)
 	s.NoError(err)
 
 	err = service.RespondDecisionTaskCompleted(ctx, completionRequest)

internal_task_handlers_test.go

@@ -117,6 +117,17 @@ func createTestEventDecisionTaskStarted(eventID int64) *s.HistoryEvent {
 		EventType: common.EventTypePtr(s.EventType_DecisionTaskStarted)}
 }
 
+func createTestEventWorkflowExecutionSignaled(eventID int64, signalName string) *s.HistoryEvent {
+	return &s.HistoryEvent{
+		EventId:   common.Int64Ptr(eventID),
+		EventType: common.EventTypePtr(s.EventType_WorkflowExecutionSignaled),
+		WorkflowExecutionSignaledEventAttributes: &s.WorkflowExecutionSignaledEventAttributes{
+			SignalName: common.StringPtr(signalName),
+			Identity:   common.StringPtr("test-identity"),
+		},
+	}
+}
+
 func createTestEventDecisionTaskCompleted(eventID int64, attr *s.DecisionTaskCompletedEventAttributes) *s.HistoryEvent {
 	return &s.HistoryEvent{
 		EventId:   common.Int64Ptr(eventID),
@@ -140,6 +151,19 @@ func createWorkflowTask(
 	}
 }
 
+func createQueryTask(
+	events []*s.HistoryEvent,
+	previousStartEventID int64,
+	workflowName string,
+	queryType string,
+) *s.PollForDecisionTaskResponse {
+	task := createWorkflowTask(events, previousStartEventID, workflowName)
+	task.Query = &s.WorkflowQuery{
+		QueryType: common.StringPtr(queryType),
+	}
+	return task
+}
+
 func (t *TaskHandlersTestSuite) TestWorkflowTask_WorkflowExecutionStarted() {
 	taskList := "tl1"
 	testEvents := []*s.HistoryEvent{
@@ -152,7 +176,8 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_WorkflowExecutionStarted() {
 		Logger:   t.logger,
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
+	request, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
+	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)
 	t.Equal(1, len(response.GetDecisions()))
@@ -181,7 +206,8 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_ActivityTaskScheduled() {
 		Logger:   t.logger,
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
+	request, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
+	response := request.(*s.RespondDecisionTaskCompletedRequest)
 
 	t.NoError(err)
 	t.NotNil(response)
@@ -191,14 +217,86 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_ActivityTaskScheduled() {
 
 	// Schedule an activity and see if we complete workflow, Having only one last decision.
 	task = createWorkflowTask(testEvents, 2, "HelloWorld_Workflow")
-	response, _, err = taskHandler.ProcessWorkflowTask(task, nil, false)
+	request, _, err = taskHandler.ProcessWorkflowTask(task, nil, false)
+	response = request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)
 	t.Equal(1, len(response.GetDecisions()))
 	t.Equal(s.DecisionType_CompleteWorkflowExecution, response.GetDecisions()[0].GetDecisionType())
 	t.NotNil(response.GetDecisions()[0].GetCompleteWorkflowExecutionDecisionAttributes())
 }
 
+func (t *TaskHandlersTestSuite) TestWorkflowTask_QueryWorkflow() {
+	// Schedule an activity and see if we complete workflow.
+	taskList := "tl1"
+	testEvents := []*s.HistoryEvent{
+		createTestEventWorkflowExecutionStarted(1, &s.WorkflowExecutionStartedEventAttributes{TaskList: &s.TaskList{Name: &taskList}}),
+		createTestEventDecisionTaskScheduled(2, &s.DecisionTaskScheduledEventAttributes{TaskList: &s.TaskList{Name: &taskList}}),
+		createTestEventDecisionTaskStarted(3),
+		createTestEventDecisionTaskCompleted(4, &s.DecisionTaskCompletedEventAttributes{ScheduledEventId: common.Int64Ptr(2)}),
+		createTestEventActivityTaskScheduled(5, &s.ActivityTaskScheduledEventAttributes{
+			ActivityId:   common.StringPtr("0"),
+			ActivityType: &s.ActivityType{Name: common.StringPtr("Greeter_Activity")},
+			TaskList:     &s.TaskList{Name: &taskList},
+		}),
+		createTestEventActivityTaskStarted(6, &s.ActivityTaskStartedEventAttributes{}),
+		createTestEventActivityTaskCompleted(7, &s.ActivityTaskCompletedEventAttributes{ScheduledEventId: common.Int64Ptr(5)}),
+		createTestEventDecisionTaskStarted(8),
+		createTestEventWorkflowExecutionSignaled(9, "test-signal"),
+	}
+	params := workerExecutionParameters{
+		TaskList: taskList,
+		Identity: "test-id-1",
+		Logger:   t.logger,
+	}
+
+	// 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(task, nil, false)
+	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(task, nil, false)
+	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(task, nil, false)
+	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(task, nil, false)
+	t.verifyQueryResult(response, "done")
+
+	task = createQueryTask(testEvents[0:9], 9, "HelloWorld_Workflow", "invalid-query-type")
+	taskHandler = newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
+	response, _, _ = taskHandler.ProcessWorkflowTask(task, nil, false)
+	t.NotNil(response)
+	queryResp, ok := response.(*s.RespondQueryTaskCompletedRequest)
+	t.True(ok)
+	t.NotNil(queryResp.ErrorMessage)
+	t.Contains(*queryResp.ErrorMessage, "unkonwn queryType")
+}
+
+func (t *TaskHandlersTestSuite) verifyQueryResult(response interface{}, expectedResult string) {
+	t.NotNil(response)
+	queryResp, ok := response.(*s.RespondQueryTaskCompletedRequest)
+	t.True(ok)
+	t.Nil(queryResp.ErrorMessage)
+	t.NotNil(queryResp.QueryResult_)
+	encodedValue := EncodedValue(queryResp.QueryResult_)
+	var queryResult string
+	err := encodedValue.Get(&queryResult)
+	t.NoError(err)
+	t.Equal(expectedResult, queryResult)
+}
+
 func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicDetection() {
 	taskList := "taskList"
 	testEvents := []*s.HistoryEvent{
@@ -216,17 +314,17 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicDetection() {
 		Logger:   zap.NewNop(),
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
-
+	request, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
+	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	// there should be no error as the history events matched the decisions.
 	t.NoError(err)
 	t.NotNil(response)
 
 	// now change the history event so it does not match to decision produced via replay
 	testEvents[1].ActivityTaskScheduledEventAttributes.ActivityType.Name = common.StringPtr("some-other-activity")
-	response, _, err = taskHandler.ProcessWorkflowTask(task, nil, false)
+	request, _, err = taskHandler.ProcessWorkflowTask(task, nil, false)
 	t.Error(err)
-	t.Nil(response)
+	t.Nil(request)
 	t.Contains(err.Error(), "nondeterministic")
 }
 
@@ -246,8 +344,8 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_CancelActivityBeforeSent() {
 		Logger:   t.logger,
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
-
+	request, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
+	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)
 	//t.printAllDecisions(response.GetDecisions())
@@ -275,10 +373,9 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_PressurePoints() {
 		Logger:   t.logger,
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, ppMgr, getHostEnvironment())
-	response, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
-
+	request, _, err := taskHandler.ProcessWorkflowTask(task, nil, false)
 	t.Error(err)
-	t.Nil(response)
+	t.Nil(request)
 }
 
 func (t *TaskHandlersTestSuite) TestWorkflowTask_PageToken() {
@@ -304,8 +401,8 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_PageToken() {
 		return &s.History{nextEvents}, nil, nil
 	}
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, getHostEnvironment())
-	response, _, err := taskHandler.ProcessWorkflowTask(task, iteratorfn, false)
-
+	request, _, err := taskHandler.ProcessWorkflowTask(task, iteratorfn, false)
+	response := request.(*s.RespondDecisionTaskCompletedRequest)
 	t.NoError(err)
 	t.NotNil(response)
 }