GetDeploymentReachability API

common/searchattribute/defs.go

@@ -47,7 +47,6 @@ const (
 	StateTransitionCount  = "StateTransitionCount"
 	TemporalChangeVersion = "TemporalChangeVersion"
 	BinaryChecksums       = "BinaryChecksums"
-	BuildIds              = "BuildIds"
 	BatcherNamespace      = "BatcherNamespace"
 	BatcherUser           = "BatcherUser"
 	HistorySizeBytes      = "HistorySizeBytes"
@@ -81,6 +80,9 @@ const (
 	// define a custom ScheduleId search attribute, in which case the query using the ScheduleId would operate just like
 	// any other custom search attribute.
 	ScheduleID = "ScheduleId"
+
+	// Used for Worker Versioning
+	BuildIds = "BuildIds"
 )
 
 var (

common/worker_versioning/worker_versioning.go

@@ -32,6 +32,7 @@ import (
 	"github.com/temporalio/sqlparser"
 	commonpb "go.temporal.io/api/common/v1"
 	deploymentpb "go.temporal.io/api/deployment/v1"
+	enumspb "go.temporal.io/api/enums/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
 	"go.temporal.io/server/common/namespace"
@@ -41,14 +42,44 @@ import (
 )
 
 const (
-	buildIdSearchAttributePrefixAssigned    = "assigned"
-	buildIdSearchAttributePrefixVersioned   = "versioned"
-	buildIdSearchAttributePrefixUnversioned = "unversioned"
-	BuildIdSearchAttributeDelimiter         = ":"
+	buildIdSearchAttributePrefixReachability = "reachability"
+	buildIdSearchAttributePrefixAssigned     = "assigned"
+	buildIdSearchAttributePrefixVersioned    = "versioned"
+	buildIdSearchAttributePrefixUnversioned  = "unversioned"
+	BuildIdSearchAttributeDelimiter          = ":"
 	// UnversionedSearchAttribute is the sentinel value used to mark all unversioned workflows
 	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)
+	return s
+}
+
+// ReachabilityBuildIdSearchAttribute returns the search attribute value for the currently assigned build ID in the form
+// 'reachability:<behavior>:<deployment_series_name>:<deployment_build_id>'
+func ReachabilityBuildIdSearchAttribute(behavior enumspb.VersioningBehavior, deployment *deploymentpb.Deployment) string {
+	var escapedDeployment string
+	if deployment == nil {
+		escapedDeployment = "UNVERSIONED"
+	} else {
+		escapedDeployment = fmt.Sprintf("%s%s%s",
+			escapeBuildIdSearchAttributeDelimiter(deployment.GetSeriesName()),
+			BuildIdSearchAttributeDelimiter,
+			escapeBuildIdSearchAttributeDelimiter(deployment.GetBuildId()),
+		)
+	}
+	return sqlparser.String(sqlparser.NewStrVal([]byte(fmt.Sprintf("%s%s%s%s%s",
+		buildIdSearchAttributePrefixReachability,
+		BuildIdSearchAttributeDelimiter,
+		escapeBuildIdSearchAttributeDelimiter(behavior.String()),
+		BuildIdSearchAttributeDelimiter,
+		escapedDeployment,
+	))))
+}
+
 // AssignedBuildIdSearchAttribute returns the search attribute value for the currently assigned build ID
 func AssignedBuildIdSearchAttribute(buildId string) string {
 	return buildIdSearchAttributePrefixAssigned + BuildIdSearchAttributeDelimiter + buildId
@@ -144,7 +175,7 @@ func DeploymentToString(deployment *deploymentpb.Deployment) string {
 	if deployment == nil {
 		return "UNVERSIONED"
 	}
-	return deployment.SeriesName + ":" + deployment.GetBuildId()
+	return deployment.GetSeriesName() + ":" + deployment.GetBuildId()
 }
 
 // MakeDirectiveForWorkflowTask returns a versioning directive based on the following parameters:

service/frontend/workflow_handler.go

@@ -3231,8 +3231,39 @@ func (wh *WorkflowHandler) decodeDeploymentMemo(memo *commonpb.Memo) *deployspb.
 	return &workflowMemo
 }
 
-func (wh *WorkflowHandler) GetDeploymentReachability(context.Context, *workflowservice.GetDeploymentReachabilityRequest) (*workflowservice.GetDeploymentReachabilityResponse, error) {
-	panic("Not implemented *yet*")
+func (wh *WorkflowHandler) GetDeploymentReachability(
+	ctx context.Context,
+	request *workflowservice.GetDeploymentReachabilityRequest,
+) (_ *workflowservice.GetDeploymentReachabilityResponse, retError error) {
+	defer log.CapturePanic(wh.logger, &retError)
+
+	if request == nil {
+		return nil, errRequestNotSet
+	}
+
+	if len(request.Namespace) == 0 {
+		return nil, errNamespaceNotSet
+	}
+
+	if !wh.config.EnableDeployments(request.Namespace) {
+		return nil, errDeploymentsNotAllowed
+	}
+
+	if request.GetDeployment() == nil {
+		return nil, serviceerror.NewInvalidArgument("deployment is required")
+	}
+
+	namespaceEntry, err := wh.namespaceRegistry.GetNamespace(namespace.Name(request.GetNamespace()))
+	if err != nil {
+		return nil, err
+	}
+	resp, err := wh.deploymentStoreClient.GetDeploymentReachability(ctx, namespaceEntry, request.Deployment.SeriesName, request.Deployment.BuildId)
+	if err != nil {
+		wh.logger.Error("Error during GetDeploymentReachability", tag.Error(err))
+		return nil, err
+	}
+
+	return resp, nil
 }
 
 func (wh *WorkflowHandler) SetCurrentDeployment(context.Context, *workflowservice.SetCurrentDeploymentRequest) (*workflowservice.SetCurrentDeploymentResponse, error) {

service/history/workflow/mutable_state_impl.go

@@ -29,6 +29,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"math/rand"
 	"reflect"
 	"slices"
@@ -2639,6 +2640,12 @@ func (ms *MutableStateImpl) UpdateBuildIdAssignment(buildId string) error {
 	return ms.updateBuildIdsSearchAttribute(&commonpb.WorkerVersionStamp{UseVersioning: true, BuildId: buildId}, limit)
 }
 
+// For v3 versioned workflows (ms.GetEffectiveVersioningBehavior() != UNSPECIFIED), this will append a tag formed as
+// reachability:<effective_behavior>:<deployment_series_name>:<deployment_build_id> to the BuildIds search attribute,
+// if it does not already exist there. The deployment will be execution_info.deployment, or the override deployment if
+// a pinned override is set.
+// For all other workflows (ms.GetEffectiveVersioningBehavior() == UNSPECIFIED), this will append a tag based on the
+// workflow's versioning status.
 func (ms *MutableStateImpl) updateBuildIdsSearchAttribute(stamp *commonpb.WorkerVersionStamp, maxSearchAttributeValueSize int) error {
 	changed, err := ms.addBuildIdToSearchAttributesWithNoVisibilityTask(stamp, maxSearchAttributeValueSize)
 	if err != nil {
@@ -2671,34 +2678,55 @@ func (ms *MutableStateImpl) loadBuildIds() ([]string, error) {
 	return searchAttributeValues, nil
 }
 
+// getReachabilityDeployment ignores DeploymentTransition.Deployment. If there is a pinned override,
+// it returns the override deployment. If there is an unpinned override or no override, it returns
+// execution_info.deployment, which is the last deployment that this workflow completed a task on.
+func (ms *MutableStateImpl) getReachabilityDeployment() *deploymentpb.Deployment {
+	versioningInfo := ms.GetExecutionInfo().GetVersioningInfo()
+	if override := versioningInfo.GetVersioningOverride(); override != nil {
+		if override.GetBehavior() == enumspb.VERSIONING_BEHAVIOR_PINNED {
+			return override.GetDeployment()
+		}
+	}
+	return versioningInfo.GetDeployment()
+}
+
 // Takes a list of loaded build IDs from a search attribute and adds a new build ID to it. Also makes sure that the
 // resulting SA list begins with either "unversioned" or "assigned:<bld>" based on workflow's Build ID assignment status.
 // Returns a potentially modified list.
 // [cleanup-old-wv] old versioning does not add "assigned:<bld>" value to the SA.
+// [cleanup-versioning-2] versioning-2 adds "assigned:<bld>" which is no longer used in versioning-3
 func (ms *MutableStateImpl) addBuildIdToLoadedSearchAttribute(
 	existingValues []string,
 	stamp *commonpb.WorkerVersionStamp,
 ) []string {
 	var newValues []string
-	if !stamp.GetUseVersioning() {
+	if !stamp.GetUseVersioning() && ms.GetEffectiveVersioningBehavior() == enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED { // unversioned workflows may still have non-nil deployment, so we don't check deployment
 		newValues = append(newValues, worker_versioning.UnversionedSearchAttribute)
+	} else if ms.GetEffectiveVersioningBehavior() != enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {
+		newValues = append(newValues, worker_versioning.ReachabilityBuildIdSearchAttribute(
+			ms.GetEffectiveVersioningBehavior(),
+			ms.getReachabilityDeployment(),
+		))
 	} else if ms.GetAssignedBuildId() != "" {
 		newValues = append(newValues, worker_versioning.AssignedBuildIdSearchAttribute(ms.GetAssignedBuildId()))
 	}
 
-	buildId := worker_versioning.VersionStampToBuildIdSearchAttribute(stamp)
-	found := slices.Contains(newValues, buildId)
-	for _, existingValue := range existingValues {
-		if existingValue == buildId {
-			found = true
+	if ms.GetEffectiveVersioningBehavior() == enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {
+		buildId := worker_versioning.VersionStampToBuildIdSearchAttribute(stamp)
+		found := slices.Contains(newValues, buildId)
+		for _, existingValue := range existingValues {
+			if existingValue == buildId {
+				found = true
+			}
+			if !worker_versioning.IsUnversionedOrAssignedBuildIdSearchAttribute(existingValue) {
+				newValues = append(newValues, existingValue)
+			}
 		}
-		if !worker_versioning.IsUnversionedOrAssignedBuildIdSearchAttribute(existingValue) {
-			newValues = append(newValues, existingValue)
+		if !found {
+			newValues = append(newValues, buildId)
 		}
 	}
-	if !found {
-		newValues = append(newValues, buildId)
-	}
 	return newValues
 }
 
@@ -4251,23 +4279,16 @@ func (ms *MutableStateImpl) AddWorkflowExecutionOptionsUpdatedEvent(
 
 func (ms *MutableStateImpl) ApplyWorkflowExecutionOptionsUpdatedEvent(event *historypb.HistoryEvent) error {
 	override := event.GetWorkflowExecutionOptionsUpdatedEventAttributes().GetVersioningOverride()
-	previousCurrentDeployment := ms.GetEffectiveDeployment()
+	previousEffectiveDeployment := ms.GetEffectiveDeployment()
 	previousEffectiveVersioningBehavior := ms.GetEffectiveVersioningBehavior()
 	if ms.GetExecutionInfo().GetVersioningInfo() == nil {
 		ms.GetExecutionInfo().VersioningInfo = &workflowpb.WorkflowExecutionVersioningInfo{}
 	}
 	ms.GetExecutionInfo().VersioningInfo.VersioningOverride = override
 
-	// If effective deployment or behavior change, we need to reschedule any pending tasks, because History will reject
-	// the task's start request if the task is being started by a poller that is not from the workflow's effective
-	// deployment according to History. Therefore, it is important for matching to match tasks with the correct pollers.
-	// Even if the effective deployment does not change, we still need to reschedule tasks into the appropriate
-	// default/unpinned queue or the pinned queue, because the two queues will be handled differently if the task queue's
-	// Current Deployment changes between now and when the task is started.
-	//
-	// We choose to let any started WFT that is running on the old deployment finish running, instead of forcing it to fail.
-	if !proto.Equal(ms.GetEffectiveDeployment(), previousCurrentDeployment) ||
+	if !proto.Equal(ms.GetEffectiveDeployment(), previousEffectiveDeployment) ||
 		ms.GetEffectiveVersioningBehavior() != previousEffectiveVersioningBehavior {
+		// TODO (carly) part 2: if safe mode, do replay test on new deployment if deployment changed, if fail, revert changes and abort
 		// If there is an ongoing transition, we remove it so that tasks from this workflow (including the pending WFT
 		// that initiated the transition) can run on our override deployment as soon as possible.
 		//
@@ -4291,21 +4312,33 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionOptionsUpdatedEvent(event *his
 		// and it will start the same transition in the workflow. So removing the transition would not make a difference
 		// and would in fact add some extra work for the server.
 		ms.executionInfo.GetVersioningInfo().DeploymentTransition = nil
-		// TODO (carly) part 2: if safe mode, do replay test on new deployment if deployment changed, if fail, revert changes and abort
+
+		// If effective deployment or behavior change, we need to reschedule any pending tasks, because History will reject
+		// the task's start request if the task is being started by a poller that is not from the workflow's effective
+		// deployment according to History. Therefore, it is important for matching to match tasks with the correct pollers.
+		// Even if the effective deployment does not change, we still need to reschedule tasks into the appropriate
+		// default/unpinned queue or the pinned queue, because the two queues will be handled differently if the task queue's
+		// Current Deployment changes between now and when the task is started.
+		//
+		// We choose to let any started WFT that is running on the old deployment finish running, instead of forcing it to fail.
 		ms.ClearStickyTaskQueue()
 		if ms.HasPendingWorkflowTask() && !ms.HasStartedWorkflowTask() &&
 			// Speculative WFT is directly (without transfer task) added to matching when scheduled.
 			// It is protected by timeout on both normal and sticky task queues.
 			// If there is no poller for previous deployment, it will time out,
 			// and will be rescheduled as normal WFT.
 			ms.GetPendingWorkflowTask().Type != enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE {
-			// sticky queue was just cleared, so the following call only generates
-			// a WorkflowTask, not a WorkflowTaskTimeoutTask.
+			// Sticky queue was just cleared, so the following call only generates a WorkflowTask, not a WorkflowTaskTimeoutTask.
 			err := ms.taskGenerator.GenerateScheduleWorkflowTaskTasks(ms.GetPendingWorkflowTask().ScheduledEventID)
 			if err != nil {
 				return err
 			}
 		}
+		// For v3 versioned workflows (ms.GetEffectiveVersioningBehavior() != UNSPECIFIED), this will update the reachability
+		// search attribute based on the execution_info.deployment and/or override deployment if one exists.
+		if err := ms.updateBuildIdsSearchAttribute(nil, math.MaxInt32); err != nil {
+			return err
+		}
 	}
 	return ms.reschedulePendingActivities()
 }

service/history/workflow/workflow_task_state_machine.go

@@ -1098,8 +1098,11 @@ func (m *workflowTaskStateMachine) afterAddWorkflowTaskCompletedEvent(
 		event.GetEventId(),
 		limits.MaxResetPoints,
 	)
-	// For versioned workflows the search attributes should be already up-to-date based on the task started events.
-	// This is still useful for unversioned workers.
+	// For v3 versioned workflows (ms.GetEffectiveVersioningBehavior() != UNSPECIFIED), this will update the reachability
+	// search attribute based on the execution_info.deployment and/or override deployment if one exists. We must update the
+	// search attribute here because the reachability deployment may have just been changed by CompleteDeploymentTransition.
+	// This is also useful for unversioned workers.
+	// For v1 and v2 versioned workflows the search attributes should be already up-to-date based on the task started events.
 	if err := m.ms.updateBuildIdsSearchAttribute(attrs.GetWorkerVersion(), limits.MaxSearchAttributeValueSize); err != nil {
 		return err
 	}

service/worker/deployment/deployment_client.go

@@ -78,6 +78,13 @@ type DeploymentStoreClient interface {
 		seriesName string,
 		NextPageToken []byte,
 	) ([]*deploymentpb.DeploymentListInfo, []byte, error)
+
+	GetDeploymentReachability(
+		ctx context.Context,
+		namespaceEntry *namespace.Namespace,
+		seriesName string,
+		buildID string,
+	) (*workflowservice.GetDeploymentReachabilityResponse, error)
 }
 
 // implements DeploymentStoreClient
@@ -86,6 +93,8 @@ type DeploymentClientImpl struct {
 	VisibilityManager     manager.VisibilityManager
 	MaxIDLengthLimit      dynamicconfig.IntPropertyFn
 	VisibilityMaxPageSize dynamicconfig.IntPropertyFnWithNamespaceFilter
+
+	reachabilityCache reachabilityCache
 }
 
 var _ DeploymentStoreClient = (*DeploymentClientImpl)(nil)
@@ -238,8 +247,39 @@ func (d *DeploymentClientImpl) DescribeDeployment(ctx context.Context, namespace
 	}, nil
 }
 
-func (d *DeploymentClientImpl) GetCurrentDeployment(ctx context.Context, namespaceEntry *namespace.Namespace, seriesName string) (*deploymentpb.DeploymentInfo, error) {
+// TODO (carly): pass deployment instead of seriesName + buildId in all these APIs -- separate PR
+func (d *DeploymentClientImpl) GetDeploymentReachability(
+	ctx context.Context,
+	namespaceEntry *namespace.Namespace,
+	seriesName string,
+	buildID string,
+) (*workflowservice.GetDeploymentReachabilityResponse, error) {
+	deployInfo, err := d.DescribeDeployment(ctx, namespaceEntry, seriesName, buildID)
+	if err != nil {
+		return nil, err
+	}
+	reachability, lastUpdateTime, err := getDeploymentReachability(
+		ctx,
+		namespaceEntry.ID().String(),
+		namespaceEntry.Name().String(),
+		seriesName,
+		buildID,
+		deployInfo.GetIsCurrent(),
+		d.reachabilityCache,
+	)
 
+	if err != nil {
+		return nil, err
+	}
+
+	return &workflowservice.GetDeploymentReachabilityResponse{
+		DeploymentInfo: deployInfo,
+		Reachability:   reachability,
+		LastUpdateTime: timestamppb.New(lastUpdateTime),
+	}, nil
+}
+
+func (d *DeploymentClientImpl) GetCurrentDeployment(ctx context.Context, namespaceEntry *namespace.Namespace, seriesName string) (*deploymentpb.DeploymentInfo, error) {
 	// Validating params
 	err := ValidateDeploymentWfParams(SeriesFieldName, seriesName, d.MaxIDLengthLimit())
 	if err != nil {