Add Functional Tests and Fix Bugs

common/dynamicconfig/constants.go

@@ -484,7 +484,9 @@ Deleted Redirect Rules will be kept in the DB (with DeleteTimestamp). After this
 		"matching.wv.ReachabilityBuildIdVisibilityGracePeriod",
 		3*time.Minute,
 		`ReachabilityBuildIdVisibilityGracePeriod is the time period for which deleted versioning rules are still considered active
-to account for the delay in updating the build id field in visibility.`,
+to account for the delay in updating the build id field in visibility. Not yet supported for GetDeploymentReachability. We recommend waiting 
+at least 2 minutes between changing the current deployment and calling GetDeployment, so that newly started workflow executions using the 
+recently-current deployment can arrive in visibility.`,
 	)
 	ReachabilityTaskQueueScanLimit = NewGlobalIntSetting(
 		"limit.reachabilityTaskQueueScan",

common/rpc/interceptor/redirection.go

@@ -128,6 +128,12 @@ var (
 		"DescribeBatchOperation":    func() any { return &workflowservice.DescribeBatchOperationResponse{} },
 		"ListBatchOperations":       func() any { return &workflowservice.ListBatchOperationsResponse{} },
 		"UpdateActivityOptionsById": func() any { return &workflowservice.UpdateActivityOptionsByIdResponse{} },
+
+		"DescribeDeployment":        func() any { return &workflowservice.DescribeDeploymentResponse{} },
+		"ListDeployments":           func() any { return &workflowservice.ListDeploymentsResponse{} },
+		"GetDeploymentReachability": func() any { return &workflowservice.GetDeploymentReachabilityResponse{} },
+		"GetCurrentDeployment":      func() any { return &workflowservice.GetCurrentDeploymentResponse{} },
+		"SetCurrentDeployment":      func() any { return &workflowservice.SetCurrentDeploymentResponse{} },
 	}
 )
 

common/worker_versioning/worker_versioning.go

@@ -51,7 +51,6 @@ const (
 	UnversionedSearchAttribute = buildIdSearchAttributePrefixUnversioned
 )
 
-// TODO (carly): fix delimiter
 // escapeBuildIdSearchAttributeDelimiter is a helper which escapes the BuildIdSearchAttributeDelimiter character in the input string
 func escapeBuildIdSearchAttributeDelimiter(s string) string {
 	s = strings.Replace(s, BuildIdSearchAttributeDelimiter, `|`+BuildIdSearchAttributeDelimiter, -1)
@@ -63,16 +62,13 @@ func escapeBuildIdSearchAttributeDelimiter(s string) string {
 // BuildIds KeywordList in this format. If the workflow becomes unpinned or unversioned, this entry will be removed from
 // that list.
 func PinnedBuildIdSearchAttribute(deployment *deploymentpb.Deployment) string {
-	escapedDeployment := fmt.Sprintf("%s%s%s",
+	return fmt.Sprintf("%s%s%s%s%s",
+		BuildIdSearchAttributePrefixPinned,
+		BuildIdSearchAttributeDelimiter,
 		escapeBuildIdSearchAttributeDelimiter(deployment.GetSeriesName()),
 		BuildIdSearchAttributeDelimiter,
 		escapeBuildIdSearchAttributeDelimiter(deployment.GetBuildId()),
 	)
-	return sqlparser.String(sqlparser.NewStrVal([]byte(fmt.Sprintf("%s%s%s",
-		BuildIdSearchAttributePrefixPinned,
-		BuildIdSearchAttributeDelimiter,
-		escapedDeployment,
-	))))
 }
 
 // AssignedBuildIdSearchAttribute returns the search attribute value for the currently assigned build ID

service/history/api/updateworkflowoptions/api.go

@@ -27,7 +27,6 @@ package updateworkflowoptions
 import (
 	"context"
 	"fmt"
-
 	"google.golang.org/protobuf/types/known/fieldmaskpb"
 
 	"go.temporal.io/api/serviceerror"
@@ -64,24 +63,22 @@ func Invoke(
 			req.GetWorkflowExecution().GetWorkflowId(),
 			req.GetWorkflowExecution().GetRunId(),
 		),
-		func(workflowLease api.WorkflowLease) (_ *api.UpdateWorkflowAction, updateError error) {
+		func(workflowLease api.WorkflowLease) (*api.UpdateWorkflowAction, error) {
 			mutableState := workflowLease.GetMutableState()
-			defer func() { workflowLease.GetReleaseFn()(updateError) }()
-
 			if !mutableState.IsWorkflowExecutionRunning() {
 				// in-memory mutable state is still clean, let updateError=nil in the defer func()
 				// to prevent clearing and reloading mutable state while releasing the lock
 				return nil, consts.ErrWorkflowCompleted
 			}
 
 			// Merge the requested options mentioned in the field mask with the current options in the mutable state
-			mergedOpts, updateError := applyWorkflowExecutionOptions(
+			mergedOpts, err := applyWorkflowExecutionOptions(
 				getOptionsFromMutableState(mutableState),
 				req.GetWorkflowExecutionOptions(),
 				req.GetUpdateMask(),
 			)
-			if updateError != nil {
-				return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("error parsing update_options: %v", updateError))
+			if err != nil {
+				return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("error applying update_options: %v", err))
 			}
 
 			// Set options for gRPC response
@@ -95,9 +92,9 @@ func Invoke(
 				}, nil
 			}
 
-			_, updateError = mutableState.AddWorkflowExecutionOptionsUpdatedEvent(req.GetWorkflowExecutionOptions().GetVersioningOverride())
-			if updateError != nil {
-				return nil, updateError
+			_, err = mutableState.AddWorkflowExecutionOptionsUpdatedEvent(mergedOpts.GetVersioningOverride())
+			if err != nil {
+				return nil, err
 			}
 
 			// TODO (carly) part 2: handle safe deployment change --> CreateWorkflowTask=true

service/history/workflow/mutable_state_impl.go

@@ -2740,7 +2740,10 @@ func (ms *MutableStateImpl) saveBuildIds(buildIds []string, maxSearchAttributeVa
 		ms.executionInfo.SearchAttributes = searchAttributes
 	}
 
-	hasUnversionedOrAssigned := worker_versioning.IsUnversionedOrAssignedBuildIdSearchAttribute(buildIds[0])
+	hasUnversionedOrAssigned := false
+	if len(buildIds) > 0 { // len is 0 if we are removing the pinned search attribute and the workflow was never unversioned or assigned
+		hasUnversionedOrAssigned = worker_versioning.IsUnversionedOrAssignedBuildIdSearchAttribute(buildIds[0])
+	}
 	for {
 		saPayload, err := searchattribute.EncodeValue(buildIds, enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST)
 		if err != nil {

service/worker/deployment/deployment_client.go

@@ -93,8 +93,7 @@ type DeploymentClientImpl struct {
 	VisibilityManager     manager.VisibilityManager
 	MaxIDLengthLimit      dynamicconfig.IntPropertyFn
 	VisibilityMaxPageSize dynamicconfig.IntPropertyFnWithNamespaceFilter
-
-	reachabilityCache reachabilityCache
+	reachabilityCache     reachabilityCache
 }
 
 var _ DeploymentStoreClient = (*DeploymentClientImpl)(nil)

service/worker/deployment/deployment_reachability.go

@@ -59,7 +59,7 @@ func getDeploymentReachability(
 ) (enumspb.DeploymentReachability, time.Time, error) {
 	// 1a. Reachable by new unpinned workflows
 	if isCurrent {
-		// TODO (carly): still return reachable if the deployment just became not current, but workflows started on it are not yet in reachability
+		// TODO (carly) part 2: still return reachable if the deployment just became not current, but workflows started on it are not yet in reachability
 		return enumspb.DEPLOYMENT_REACHABILITY_REACHABLE, time.Now(), nil
 	}
 
@@ -99,10 +99,10 @@ func makeCountRequest(
 
 func makeDeploymentQuery(seriesName, buildID string, open bool) string {
 	var statusFilter string
-	deploymentFilter := "= " + worker_versioning.PinnedBuildIdSearchAttribute(&deploymentpb.Deployment{
+	deploymentFilter := fmt.Sprintf("= '%s'", worker_versioning.PinnedBuildIdSearchAttribute(&deploymentpb.Deployment{
 		SeriesName: seriesName,
 		BuildId:    buildID,
-	})
+	}))
 	if open {
 		statusFilter = "= 'Running'"
 	} else {

service/worker/deployment/deployment_reachability_test.go

@@ -47,11 +47,11 @@ func TestMakeDeploymentQuery(t *testing.T) {
 	buildId := "A"
 
 	query := makeDeploymentQuery(seriesName, buildId, true)
-	expectedQuery := "BuildIds = 'reachability:Pinned:test-deployment:A' AND ExecutionStatus = 'Running'"
+	expectedQuery := "BuildIds = 'pinned:test-deployment:A' AND ExecutionStatus = 'Running'"
 	assert.Equal(t, expectedQuery, query)
 
 	query = makeDeploymentQuery(seriesName, buildId, false)
-	expectedQuery = "BuildIds = 'reachability:Pinned:test-deployment:A' AND ExecutionStatus != 'Running'"
+	expectedQuery = "BuildIds = 'pinned:test-deployment:A' AND ExecutionStatus != 'Running'"
 	assert.Equal(t, expectedQuery, query)
 }
 
@@ -63,9 +63,8 @@ func TestReachable_CurrentDeployment(t *testing.T) {
 	vm := manager.NewMockVisibilityManager(gomock.NewController(t)) // won't receive any calls
 	testCache := newReachabilityCache(metrics.NoopMetricsHandler, vm, testReachabilityCacheOpenWFsTTL, testReachabilityCacheClosedWFsTTL)
 
-	reach, reachValidTime, err := getDeploymentReachability(ctx, "", "", seriesName, buildId, true, testCache)
+	reach, _, err := getDeploymentReachability(ctx, "", "", seriesName, buildId, true, testCache)
 	assert.Nil(t, err)
-	assert.Greater(t, time.Now(), reachValidTime)
 	assert.Equal(t, enumspb.DEPLOYMENT_REACHABILITY_REACHABLE, reach)
 }
 

service/worker/deployment/fx.go

@@ -95,8 +95,8 @@ func DeploymentStoreClientProvider(historyClient resource.HistoryClient, visibil
 		reachabilityCache: newReachabilityCache(
 			metrics.NoopMetricsHandler,
 			visibilityManager,
-			reachabilityCacheOpenWFsTTL,   // TODO (carly) use dc (ie. config.ReachabilityCacheOpenWFsTTL)
-			reachabilityCacheClosedWFsTTL, // TODO (carly) use dc (ie. config.ReachabilityCacheClosedWFsTTL)
+			dynamicconfig.ReachabilityCacheOpenWFsTTL.Get(dc)(),
+			dynamicconfig.ReachabilityCacheClosedWFsTTL.Get(dc)(),
 		),
 	}
 }