Consistent query test (#916)

Author: andrewjdawson2016

internal/internal_task_pollers.go

@@ -368,6 +368,8 @@ func (wtp *workflowTaskPoller) RespondTaskCompleted(completedRequest interface{}
 						WorkerTaskList:                &s.TaskList{Name: common.StringPtr(getWorkerTaskList(wtp.stickyUUID))},
 						ScheduleToStartTimeoutSeconds: common.Int32Ptr(common.Int32Ceil(wtp.StickyScheduleToStartTimeout.Seconds())),
 					}
+				} else {
+					request.ReturnNewDecisionTask = common.BoolPtr(false)
 				}
 				response, err1 = wtp.service.RespondDecisionTaskCompleted(tchCtx, request, opt...)
 				if err1 != nil {

internal/version.go

@@ -42,4 +42,4 @@ const LibraryVersion = "0.10.1"
 // This can be used for client capability check, on
 // Cadence server, for backward compatibility
 // Format: MAJOR.MINOR.PATCH
-const FeatureVersion = "1.4.0"
+const FeatureVersion = "1.5.0"

test/activity_test.go

@@ -48,6 +48,11 @@ func (a *Activities) Sleep(ctx context.Context, delay time.Duration) error {
 	return nil
 }
 
+func LocalSleep(ctx context.Context, delay time.Duration) error {
+	time.Sleep(delay)
+	return nil
+}
+
 func (a *Activities) HeartbeatAndSleep(ctx context.Context, seq int, delay time.Duration) (int, error) {
 	a.append("heartbeatAndSleep")
 	if activity.HasHeartbeatDetails(ctx) {

test/integration_test.go

@@ -239,6 +239,39 @@ func (ts *IntegrationTestSuite) TestStackTraceQuery() {
 	ts.True(strings.Contains(trace, "go.uber.org/cadence/test.(*Workflows).Basic"))
 }
 
+func (ts *IntegrationTestSuite) TestConsistentQuery() {
+	ctx, cancel := context.WithTimeout(context.Background(), ctxTimeout)
+	defer cancel()
+	// this workflow will start a local activity which blocks for long enough
+	// to ensure that consistent query must wait in order to satisfy consistency
+	wfOpts := ts.startWorkflowOptions("test-consistent-query")
+	wfOpts.DecisionTaskStartToCloseTimeout = 5 *time.Second
+	run, err := ts.libClient.ExecuteWorkflow(ctx, wfOpts, ts.workflows.ConsistentQueryWorkflow, 3*time.Second)
+	ts.Nil(err)
+	// Wait for a second to ensure that first decision task gets started and completed before we send signal.
+	// Query cannot be run until first decision task has been completed.
+	// If signal occurs right after workflow start then WorkflowStarted and Signal events will both be part of the same
+	// decision task. So query will be blocked waiting for signal to complete, this is not what we want because it
+	// will not exercise the consistent query code path.
+	<-time.After(time.Second)
+	err = ts.libClient.SignalWorkflow(ctx, "test-consistent-query", run.GetRunID(), consistentQuerySignalCh, "signal-input")
+	ts.NoError(err)
+
+	value, err := ts.libClient.QueryWorkflowWithOptions(ctx, &client.QueryWorkflowWithOptionsRequest{
+		WorkflowID: "test-consistent-query",
+		RunID: run.GetRunID(),
+		QueryType: "consistent_query",
+		QueryConsistencyLevel: shared.QueryConsistencyLevelStrong.Ptr(),
+	})
+	ts.Nil(err)
+	ts.NotNil(value)
+	ts.NotNil(value.QueryResult)
+	ts.Nil(value.QueryRejected)
+	var queryResult string
+	ts.Nil(value.QueryResult.Get(&queryResult))
+	ts.Equal("signal-input", queryResult)
+}
+
 func (ts *IntegrationTestSuite) TestWorkflowIDReuseRejectDuplicate() {
 	var result string
 	err := ts.executeWorkflow(

test/workflow_test.go

@@ -33,6 +33,10 @@ import (
 	"go.uber.org/cadence/workflow"
 )
 
+const (
+	consistentQuerySignalCh = "consistent-query-signal-chan"
+)
+
 type Workflows struct{}
 
 func (w *Workflows) Basic(ctx workflow.Context) ([]string, error) {
@@ -409,6 +413,25 @@ func (w *Workflows) LargeQueryResultWorkflow(ctx workflow.Context) (string, erro
 	return "hello", nil
 }
 
+func (w *Workflows) ConsistentQueryWorkflow(ctx workflow.Context, delay time.Duration) error {
+	queryResult := "starting-value"
+	err := workflow.SetQueryHandler(ctx, "consistent_query", func() (string, error) {
+		return queryResult, nil
+	})
+	if err != nil {
+		return errors.New("failed to register query handler")
+	}
+	ch := workflow.GetSignalChannel(ctx, consistentQuerySignalCh)
+	var signalData string
+	ch.Receive(ctx, &signalData)
+	laCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
+		ScheduleToCloseTimeout: 5 * time.Second,
+	})
+	workflow.ExecuteLocalActivity(laCtx, LocalSleep, delay).Get(laCtx, nil)
+	queryResult = signalData
+	return nil
+}
+
 func (w *Workflows) RetryTimeoutStableErrorWorkflow(ctx workflow.Context) ([]string, error) {
 	ao := workflow.ActivityOptions{
 		ScheduleToStartTimeout: time.Second * 2,
@@ -486,6 +509,7 @@ func (w *Workflows) register() {
 	workflow.Register(w.SimplestWorkflow)
 	workflow.Register(w.LargeQueryResultWorkflow)
 	workflow.Register(w.RetryTimeoutStableErrorWorkflow)
+	workflow.Register(w.ConsistentQueryWorkflow)
 }
 
 func (w *Workflows) defaultActivityOptions() workflow.ActivityOptions {