add external type for QueryConsistencyLevel, mappers to internal thrift types

Author: c-warren

client/client.go

@@ -87,6 +87,11 @@ type (
 	// ParentClosePolicy defines the behavior performed on a child workflow when its parent is closed
 	ParentClosePolicy = internal.ParentClosePolicy
 
+	// QueryConsistencyLevel defines the level of consistency the query should respond with.
+	// It will default to the cluster's configuration if not specified.
+	// Valid values are QueryConsistencyLevelEventual (served by the receiving cluster), and QueryConsistencyLevelStrong (redirects to the active cluster).
+	QueryConsistencyLevel = internal.QueryConsistencyLevel
+
 	// CancelOption values are functional options for the CancelWorkflow method.
 	// Supported values can be created with:
 	//  - WithCancelReason(...)
@@ -492,6 +497,15 @@ const (
 	ParentClosePolicyAbandon = internal.ParentClosePolicyAbandon
 )
 
+const (
+	// QueryConsistencyLevelUnspecified will use the default consistency level provided by the cluster.
+	QueryConsistencyLevelUnspecified = internal.QueryConsistencyLevelUnspecified
+	// QueryConsistencyLevelEventual passes the request to the receiving cluster and returns eventually consistent results
+	QueryConsistencyLevelEventual = internal.QueryConsistencyLevelEventual
+	// QueryConsistencyLevelStrong will redirect the request to the active cluster and returns strongly consistent results
+	QueryConsistencyLevelStrong = internal.QueryConsistencyLevelStrong
+)
+
 // NewClient creates an instance of a workflow client
 func NewClient(service workflowserviceclient.Interface, domain string, options *Options) Client {
 	return internal.NewClient(service, domain, options)

internal/client.go

@@ -586,6 +586,11 @@ type (
 	// ParentClosePolicy defines the action on children when parent is closed
 	ParentClosePolicy int
 
+	// QueryConsistencyLevel defines the level of consistency the query should respond with.
+	// It will default to the cluster's configuration if not specified.
+	// Valid values are QueryConsistencyLevelEventual (served by the receiving cluster), and QueryConsistencyLevelStrong (redirects to the active cluster).
+	QueryConsistencyLevel int
+
 	// ActiveClusterSelectionPolicy defines the policy for selecting the active cluster to start the workflow execution on for active-active domains.
 	// Active-active domains can be configured to be active in multiple clusters (at most one in a given region).
 	// Individual workflows can be configured to be active in one of the active clusters of the domain.
@@ -637,6 +642,15 @@ const (
 	WorkflowIDReusePolicyTerminateIfRunning
 )
 
+const (
+	// QueryConsistencyLevelUnspecified will use the default consistency level provided by the cluster.
+	QueryConsistencyLevelUnspecified QueryConsistencyLevel = iota
+	// QueryConsistencyLevelEventual passes the request to the receiving cluster and returns eventually consistent results
+	QueryConsistencyLevelEventual
+	// QueryConsistencyLevelStrong will redirect the request to the active cluster and returns strongly consistent results
+	QueryConsistencyLevelStrong
+)
+
 func getFeatureFlags(options *ClientOptions) FeatureFlags {
 	if options != nil {
 		return FeatureFlags{

internal/convert.go

@@ -72,3 +72,14 @@ func convertActiveClusterSelectionPolicy(policy *ActiveClusterSelectionPolicy) (
 		return nil, fmt.Errorf("invalid active cluster selection strategy: %d", policy.Strategy)
 	}
 }
+
+func convertQueryConsistencyLevel(level QueryConsistencyLevel) *s.QueryConsistencyLevel {
+	switch level {
+	case QueryConsistencyLevelEventual:
+		return s.QueryConsistencyLevelEventual.Ptr()
+	case QueryConsistencyLevelStrong:
+		return s.QueryConsistencyLevelStrong.Ptr()
+	default:
+		return nil
+	}
+}

internal/convert_test.go

@@ -157,3 +157,39 @@ func TestConvertActiveClusterSelectionPolicy(t *testing.T) {
 		}
 	}
 }
+
+func TestConvertQueryConsistencyLevel(t *testing.T) {
+	tests := []struct {
+		name           string
+		level          QueryConsistencyLevel
+		expectedThrift *s.QueryConsistencyLevel
+	}{
+		{
+			name:           "QueryConsistencyLevelUnspecified - return nil",
+			level:          QueryConsistencyLevelUnspecified,
+			expectedThrift: nil,
+		},
+		{
+			name:           "QueryConsistencyLevelEventual - return eventual",
+			level:          QueryConsistencyLevelEventual,
+			expectedThrift: s.QueryConsistencyLevelEventual.Ptr(),
+		},
+		{
+			name:           "QueryConsistencyLevelStrong - return strong",
+			level:          QueryConsistencyLevelStrong,
+			expectedThrift: s.QueryConsistencyLevelStrong.Ptr(),
+		},
+		{
+			name:           "invalid level (999) - return nil",
+			level:          QueryConsistencyLevel(999),
+			expectedThrift: nil,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := convertQueryConsistencyLevel(test.level)
+			assert.Equal(t, test.expectedThrift, result)
+		})
+	}
+}

internal/internal_workflow_client.go

@@ -462,7 +462,7 @@ type GetWorkflowHistoryWithOptionsRequest struct {
 	// QueryConsistencyLevel is an optional field used to specify the consistency level for the query.
 	// QueryConsistencyLevelStrong will query the currently active cluster for this workflow - at the potential cost of additional latency.
 	// If not set, server will use the default consistency level.
-	QueryConsistencyLevel *s.QueryConsistencyLevel
+	QueryConsistencyLevel QueryConsistencyLevel
 }
 
 // GetWorkflowHistory return a channel which contains the history events of a given workflow
@@ -478,7 +478,7 @@ func (wc *workflowClient) GetWorkflowHistory(
 		RunID:                 runID,
 		IsLongPoll:            isLongPoll,
 		FilterType:            filterType,
-		QueryConsistencyLevel: nil, // Use server default
+		QueryConsistencyLevel: QueryConsistencyLevelUnspecified,
 	}
 	iter, _ := wc.GetWorkflowHistoryWithOptions(ctx, request)
 	return iter
@@ -503,7 +503,7 @@ func (wc *workflowClient) GetWorkflowHistoryWithOptions(ctx context.Context, req
 			HistoryEventFilterType: &request.FilterType,
 			NextPageToken:          nextToken,
 			SkipArchival:           common.BoolPtr(request.IsLongPoll),
-			QueryConsistencyLevel:  request.QueryConsistencyLevel,
+			QueryConsistencyLevel:  convertQueryConsistencyLevel(request.QueryConsistencyLevel),
 		}
 
 		var response *s.GetWorkflowExecutionHistoryResponse
@@ -830,7 +830,7 @@ type DescribeWorkflowExecutionWithOptionsRequest struct {
 	// QueryConsistencyLevel is an optional field used to specify the consistency level for the query.
 	// QueryConsistencyLevelStrong will query the currently active cluster for this workflow - at the potential cost of additional latency.
 	// If not set, server will use the default consistency level.
-	QueryConsistencyLevel *s.QueryConsistencyLevel
+	QueryConsistencyLevel QueryConsistencyLevel
 }
 
 // DescribeWorkflowExecution returns information about the specified workflow execution.
@@ -842,7 +842,7 @@ func (wc *workflowClient) DescribeWorkflowExecution(ctx context.Context, workflo
 	request := &DescribeWorkflowExecutionWithOptionsRequest{
 		WorkflowID:            workflowID,
 		RunID:                 runID,
-		QueryConsistencyLevel: nil, // Use server default
+		QueryConsistencyLevel: QueryConsistencyLevelUnspecified,
 	}
 	return wc.DescribeWorkflowExecutionWithOptions(ctx, request)
 }
@@ -860,7 +860,7 @@ func (wc *workflowClient) DescribeWorkflowExecutionWithOptions(ctx context.Conte
 			WorkflowId: common.StringPtr(request.WorkflowID),
 			RunId:      common.StringPtr(request.RunID),
 		},
-		QueryConsistencyLevel: request.QueryConsistencyLevel,
+		QueryConsistencyLevel: convertQueryConsistencyLevel(request.QueryConsistencyLevel),
 	}
 	var response *s.DescribeWorkflowExecutionResponse
 	err := backoff.Retry(ctx,

internal/internal_workflow_client_test.go

@@ -2056,7 +2056,7 @@ func (s *workflowClientTestSuite) TestDescribeWorkflowExecutionWithOptions() {
 			request: &DescribeWorkflowExecutionWithOptionsRequest{
 				WorkflowID:            workflowID,
 				RunID:                 runID,
-				QueryConsistencyLevel: shared.QueryConsistencyLevelEventual.Ptr(),
+				QueryConsistencyLevel: QueryConsistencyLevelEventual,
 			},
 			requestValidator: func(req *shared.DescribeWorkflowExecutionRequest) {
 				s.Equal(domain, req.GetDomain())
@@ -2077,7 +2077,7 @@ func (s *workflowClientTestSuite) TestDescribeWorkflowExecutionWithOptions() {
 			request: &DescribeWorkflowExecutionWithOptionsRequest{
 				WorkflowID:            workflowID,
 				RunID:                 runID,
-				QueryConsistencyLevel: shared.QueryConsistencyLevelStrong.Ptr(),
+				QueryConsistencyLevel: QueryConsistencyLevelStrong,
 			},
 			requestValidator: func(req *shared.DescribeWorkflowExecutionRequest) {
 				s.Equal(domain, req.GetDomain())
@@ -2673,7 +2673,7 @@ func (s *workflowClientTestSuite) TestGetWorkflowHistoryWithOptions() {
 				RunID:                 runID,
 				IsLongPoll:            true,
 				FilterType:            shared.HistoryEventFilterTypeCloseEvent,
-				QueryConsistencyLevel: shared.QueryConsistencyLevelEventual.Ptr(),
+				QueryConsistencyLevel: QueryConsistencyLevelEventual,
 			},
 			requestValidator: func(req *shared.GetWorkflowExecutionHistoryRequest) {
 				s.Equal(domain, req.GetDomain())
@@ -2709,7 +2709,7 @@ func (s *workflowClientTestSuite) TestGetWorkflowHistoryWithOptions() {
 				RunID:                 runID,
 				IsLongPoll:            false,
 				FilterType:            shared.HistoryEventFilterTypeAllEvent,
-				QueryConsistencyLevel: shared.QueryConsistencyLevelStrong.Ptr(),
+				QueryConsistencyLevel: QueryConsistencyLevelStrong,
 			},
 			requestValidator: func(req *shared.GetWorkflowExecutionHistoryRequest) {
 				s.Equal(domain, req.GetDomain())
@@ -2745,7 +2745,7 @@ func (s *workflowClientTestSuite) TestGetWorkflowHistoryWithOptions() {
 				RunID:                 runID,
 				IsLongPoll:            false,
 				FilterType:            shared.HistoryEventFilterTypeAllEvent,
-				QueryConsistencyLevel: shared.QueryConsistencyLevelStrong.Ptr(),
+				QueryConsistencyLevel: QueryConsistencyLevelStrong,
 			},
 			requestValidator: func(req *shared.GetWorkflowExecutionHistoryRequest) {
 				s.Equal(domain, req.GetDomain())

test/integration_test.go

@@ -320,7 +320,7 @@ func (ts *IntegrationTestSuite) TestConsistentQuery() {
 	descResp, err := ts.libClient.DescribeWorkflowExecutionWithOptions(ctx, &client.DescribeWorkflowExecutionWithOptionsRequest{
 		WorkflowID:            "test-consistent-query",
 		RunID:                 run.GetRunID(),
-		QueryConsistencyLevel: shared.QueryConsistencyLevelStrong.Ptr(),
+		QueryConsistencyLevel: client.QueryConsistencyLevelStrong,
 	})
 	ts.Nil(err)
 	ts.NotNil(descResp)
@@ -334,7 +334,7 @@ func (ts *IntegrationTestSuite) TestConsistentQuery() {
 		RunID:                 run.GetRunID(),
 		IsLongPoll:            false,
 		FilterType:            shared.HistoryEventFilterTypeAllEvent,
-		QueryConsistencyLevel: shared.QueryConsistencyLevelStrong.Ptr(),
+		QueryConsistencyLevel: client.QueryConsistencyLevelStrong,
 	})
 	ts.Nil(err)
 	ts.NotNil(histIter)