Merge branch 'master' into mandychen/remove-buildkite

Author: shijiesheng

.github/workflows/ci-checks.yml

@@ -29,6 +29,7 @@ jobs:
             docker compose -f docker/github_actions/docker-compose.yml run unit-test bash -c "make unit_test && ./scripts/gen_coverage_metadata.sh .build/metadata.txt"
 
       - name: Upload coverage artifacts
+        if: always()
         uses: actions/upload-artifact@v4
         with:
           name: go-unit-test-coverage
@@ -82,6 +83,7 @@ jobs:
             docker compose -f docker/github_actions/docker-compose.yml run integ-test bash -c "make integ_test_sticky_off"
 
       - name: Upload coverage artifacts
+        if: always()
         uses: actions/upload-artifact@v4
         with:
           name: go-integration-sticky-off-coverage
@@ -113,6 +115,7 @@ jobs:
             docker compose -f docker/github_actions/docker-compose.yml run integ-test bash -c "make integ_test_sticky_on"
 
       - name: Upload coverage artifacts
+        if: always()
         uses: actions/upload-artifact@v4
         with:
           name: go-integration-sticky-on-coverage
@@ -144,6 +147,7 @@ jobs:
             docker compose -f docker/github_actions/docker-compose.yml run integ-test-grpc bash -c "make integ_test_grpc"
 
       - name: Upload coverage artifacts
+        if: always()
         uses: actions/upload-artifact@v4
         with:
           name: go-integration-grpc-coverage

activity/activity.go

@@ -102,6 +102,11 @@ func GetInfo(ctx context.Context) Info {
 	return internal.GetActivityInfo(ctx)
 }
 
+// HasInfo returns if the context contains activity information
+func HasInfo(ctx context.Context) bool {
+	return internal.HasActivityInfo(ctx)
+}
+
 // GetLogger returns a logger that can be used in activity
 func GetLogger(ctx context.Context) *zap.Logger {
 	return internal.GetActivityLogger(ctx)

client/client.go

@@ -78,9 +78,20 @@ type (
 	// QueryWorkflowWithOptionsResponse defines the response to QueryWorkflowWithOptions
 	QueryWorkflowWithOptionsResponse = internal.QueryWorkflowWithOptionsResponse
 
+	// GetWorkflowHistoryWithOptionsRequest defines the request to GetWorkflowHistoryWithOptions
+	GetWorkflowHistoryWithOptionsRequest = internal.GetWorkflowHistoryWithOptionsRequest
+
+	// DescribeWorkflowExecutionWithOptionsRequest defines the request to DescribeWorkflowExecutionWithOptions
+	DescribeWorkflowExecutionWithOptionsRequest = internal.DescribeWorkflowExecutionWithOptionsRequest
+
 	// 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(...)
@@ -226,6 +237,26 @@ type (
 		//		}
 		GetWorkflowHistory(ctx context.Context, workflowID string, runID string, isLongPoll bool, filterType s.HistoryEventFilterType) HistoryEventIterator
 
+		// GetWorkflowHistoryWithOptions gets history events of a particular workflow.
+		// See GetWorkflowHistoryWithOptionsRequest for more information.
+		// Returns an iterator of HistoryEvents - see shared.HistoryEvent for more details.
+		// Example:-
+		//	To iterate all events,
+		// 		iter := GetWorkflowHistory(ctx, workflowID, runID, isLongPoll, filterType)
+		//		events := []*shared.HistoryEvent{}
+		//		for iter.HasNext() {
+		//			event, err := iter.Next()
+		//			if err != nil {
+		//				return err
+		//			}
+		//			events = append(events, event)
+		//		}
+		// Returns the following errors:
+		//  - EntityNotExistsError
+		//  - BadRequestError
+		//  - InternalServiceError
+		GetWorkflowHistoryWithOptions(ctx context.Context, request *GetWorkflowHistoryWithOptionsRequest) HistoryEventIterator
+
 		// CompleteActivity reports activity completed.
 		// activity Execute method can return activity.ErrResultPending to
 		// indicate the activity is not completed when it's Execute method returns. In that case, this CompleteActivity() method
@@ -383,6 +414,14 @@ type (
 		//  - EntityNotExistError
 		DescribeWorkflowExecution(ctx context.Context, workflowID, runID string) (*s.DescribeWorkflowExecutionResponse, error)
 
+		// DescribeWorkflowExecutionWithOptions returns information about the specified workflow execution.
+		// See DescribeWorkflowExecutionWithOptionsRequest for more information.
+		// The errors it can return:
+		//  - BadRequestError
+		//  - InternalServiceError
+		//  - EntityNotExistError
+		DescribeWorkflowExecutionWithOptions(ctx context.Context, request *DescribeWorkflowExecutionWithOptionsRequest) (*s.DescribeWorkflowExecutionResponse, error)
+
 		// DescribeTaskList returns information about the target tasklist, right now this API returns the
 		// pollers which polled this tasklist in last few minutes.
 		// The errors it can return:
@@ -458,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/activity.go

@@ -224,6 +224,11 @@ func GetActivityInfo(ctx context.Context) ActivityInfo {
 	}
 }
 
+// HasActivityInfo returns if the context contains activity information
+func HasActivityInfo(ctx context.Context) bool {
+	return hasActivityEnv(ctx)
+}
+
 // HasHeartbeatDetails checks if there is heartbeat details from last attempt.
 func HasHeartbeatDetails(ctx context.Context) bool {
 	env := getActivityEnv(ctx)

internal/activity_test.go

@@ -233,3 +233,25 @@ func (s *activityTestSuite) TestGetWorkerStopChannel() {
 	channel := GetWorkerStopChannel(ctx)
 	s.NotNil(channel)
 }
+
+func (s *activityTestSuite) TestHasActivityInfo() {
+	// Test context without activity info
+	ctx := context.Background()
+	s.False(HasActivityInfo(ctx))
+
+	// Test context with activity info
+	activityEnv := &activityEnvironment{
+		activityID:   "test-activity-id",
+		activityType: ActivityType{Name: "test-activity-type"},
+	}
+	ctxWithActivity := context.WithValue(ctx, activityEnvContextKey, activityEnv)
+	s.True(HasActivityInfo(ctxWithActivity))
+
+	// Test context with nil activity env
+	ctxWithNilActivity := context.WithValue(ctx, activityEnvContextKey, nil)
+	s.False(HasActivityInfo(ctxWithNilActivity))
+
+	// Test context with other values in context
+	ctxWithOtherValue := context.WithValue(ctx, activityOptionsContextKey, "other-value")
+	s.False(HasActivityInfo(ctxWithOtherValue))
+}

internal/client.go

@@ -219,6 +219,15 @@ type (
 		//		}
 		GetWorkflowHistory(ctx context.Context, workflowID string, runID string, isLongPoll bool, filterType s.HistoryEventFilterType) HistoryEventIterator
 
+		// GetWorkflowHistoryWithOptions gets history events of a particular workflow.
+		// See GetWorkflowHistoryWithOptionsRequest for more information.
+		// Returns an iterator of HistoryEvents - see shared.HistoryEvent for more details.
+		// The errors it can return:
+		//  - EntityNotExistsError
+		//  - BadRequestError
+		//  - InternalServiceError
+		GetWorkflowHistoryWithOptions(ctx context.Context, request *GetWorkflowHistoryWithOptionsRequest) HistoryEventIterator
+
 		// CompleteActivity reports activity completed.
 		// activity Execute method can return acitivity.activity.ErrResultPending to
 		// indicate the activity is not completed when it's Execute method returns. In that case, this CompleteActivity() method
@@ -367,6 +376,14 @@ type (
 		//  - EntityNotExistError
 		DescribeWorkflowExecution(ctx context.Context, workflowID, runID string) (*s.DescribeWorkflowExecutionResponse, error)
 
+		// DescribeWorkflowExecutionWithOptions returns information about the specified workflow execution.
+		// See DescribeWorkflowExecutionWithOptionsRequest for more information.
+		// The errors it can return:
+		//  - BadRequestError
+		//  - InternalServiceError
+		//  - EntityNotExistError
+		DescribeWorkflowExecutionWithOptions(ctx context.Context, request *DescribeWorkflowExecutionWithOptionsRequest) (*s.DescribeWorkflowExecutionResponse, error)
+
 		// DescribeTaskList returns information about the target tasklist, right now this API returns the
 		// pollers which polled this tasklist in last few minutes.
 		// The errors it can return:
@@ -569,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.
@@ -620,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/compatibility/proto/decision.go

@@ -142,6 +142,8 @@ func Decision(d *shared.Decision) *apiv1.Decision {
 				Memo:                         Memo(attr.Memo),
 				SearchAttributes:             SearchAttributes(attr.SearchAttributes),
 				JitterStart:                  secondsToDuration(attr.JitterStartSeconds),
+				CronOverlapPolicy:            CronOverlapPolicy(attr.CronOverlapPolicy),
+				ActiveClusterSelectionPolicy: ActiveClusterSelectionPolicy(attr.ActiveClusterSelectionPolicy),
 			},
 		}
 	case shared.DecisionTypeStartChildWorkflowExecution:
@@ -163,6 +165,8 @@ func Decision(d *shared.Decision) *apiv1.Decision {
 				Header:                       Header(attr.Header),
 				Memo:                         Memo(attr.Memo),
 				SearchAttributes:             SearchAttributes(attr.SearchAttributes),
+				CronOverlapPolicy:            CronOverlapPolicy(attr.CronOverlapPolicy),
+				ActiveClusterSelectionPolicy: ActiveClusterSelectionPolicy(attr.ActiveClusterSelectionPolicy),
 			},
 		}
 	case shared.DecisionTypeSignalExternalWorkflowExecution:

internal/compatibility/proto/request.go

@@ -519,6 +519,7 @@ const (
 	DomainUpdateVisibilityArchivalURIField    = "visibility_archival_uri"
 	DomainUpdateActiveClusterNameField        = "active_cluster_name"
 	DomainUpdateClustersField                 = "clusters"
+	DomainUpdateActiveClustersField           = "active_clusters"
 	DomainUpdateDeleteBadBinaryField          = "delete_bad_binary"
 	DomainUpdateFailoverTimeoutField          = "failover_timeout"
 )
@@ -583,6 +584,10 @@ func UpdateDomainRequest(t *shared.UpdateDomainRequest) *apiv1.UpdateDomainReque
 			request.Clusters = ClusterReplicationConfigurationArray(replicationConfiguration.Clusters)
 			fields = append(fields, DomainUpdateClustersField)
 		}
+		if replicationConfiguration.ActiveClusters != nil {
+			request.ActiveClusters = ActiveClusters(replicationConfiguration.ActiveClusters)
+			fields = append(fields, DomainUpdateActiveClustersField)
+		}
 	}
 	if t.DeleteBadBinary != nil {
 		request.DeleteBadBinary = *t.DeleteBadBinary

internal/compatibility/proto/response.go

@@ -62,6 +62,7 @@ func DescribeDomainResponse(t *shared.DescribeDomainResponse) *apiv1.DescribeDom
 	if repl := t.ReplicationConfiguration; repl != nil {
 		domain.ActiveClusterName = repl.GetActiveClusterName()
 		domain.Clusters = ClusterReplicationConfigurationArray(repl.Clusters)
+		domain.ActiveClusters = ActiveClusters(repl.ActiveClusters)
 	}
 	return &apiv1.DescribeDomainResponse{Domain: &domain}
 }
@@ -341,6 +342,7 @@ func UpdateDomainResponse(t *shared.UpdateDomainResponse) *apiv1.UpdateDomainRes
 	if repl := t.ReplicationConfiguration; repl != nil {
 		domain.ActiveClusterName = repl.GetActiveClusterName()
 		domain.Clusters = ClusterReplicationConfigurationArray(repl.Clusters)
+		domain.ActiveClusters = ActiveClusters(repl.ActiveClusters)
 	}
 	return &apiv1.UpdateDomainResponse{
 		Domain: domain,

internal/compatibility/proto/types.go

@@ -417,6 +417,7 @@ func WorkflowExecutionInfo(t *shared.WorkflowExecutionInfo) *apiv1.WorkflowExecu
 		AutoResetPoints:     ResetPoints(t.AutoResetPoints),
 		TaskList:            t.GetTaskList(),
 		IsCron:              t.GetIsCron(),
+		CronOverlapPolicy:   CronOverlapPolicy(t.CronOverlapPolicy),
 	}
 }
 
@@ -552,6 +553,7 @@ func DescribeDomainResponseDomain(t *shared.DescribeDomainResponse) *apiv1.Domai
 	if repl := t.ReplicationConfiguration; repl != nil {
 		domain.ActiveClusterName = repl.GetActiveClusterName()
 		domain.Clusters = ClusterReplicationConfigurationArray(repl.Clusters)
+		domain.ActiveClusters = ActiveClusters(repl.ActiveClusters)
 	}
 	return &domain
 }
@@ -695,3 +697,21 @@ func ActivityLocalDispatchInfoMap(t map[string]*shared.ActivityLocalDispatchInfo
 	}
 	return v
 }
+
+func ActiveClusters(ac *shared.ActiveClusters) *apiv1.ActiveClusters {
+	if ac == nil {
+		return nil
+	}
+
+	regToCl := make(map[string]*apiv1.ActiveClusterInfo)
+	for reg, clusterInfo := range ac.ActiveClustersByRegion {
+		regToCl[reg] = &apiv1.ActiveClusterInfo{
+			ActiveClusterName: clusterInfo.GetActiveClusterName(),
+			FailoverVersion:   clusterInfo.GetFailoverVersion(),
+		}
+	}
+
+	return &apiv1.ActiveClusters{
+		RegionToCluster: regToCl,
+	}
+}