Sync WCI compute status to worker deployment version state

Add DescribeWorkerControllerInstanceStatus activity to fetch the WCI validation status and map it to a ComputeStatus. During handleSyncState, invoke syncVersionDataToComputeStatus to update the version state's ComputeStatus with the provider validation result. The sync is gated behind the "sync-compute-validation-status" workflow version for backward compatibility.

Author: 02strich

service/worker/workerdeployment/version_activities.go

@@ -19,6 +19,7 @@ import (
 	"go.temporal.io/sdk/temporal"
 	deploymentspb "go.temporal.io/server/api/deployment/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/worker_versioning"
@@ -199,6 +200,19 @@ func (a *VersionActivities) GetVersionDrainageStatus(ctx context.Context, versio
 	}, nil
 }
 
+func (a *VersionActivities) DescribeWorkerControllerInstanceStatus(ctx context.Context, version *deploymentspb.WorkerDeploymentVersion) (*deploymentpb.ComputeStatus, error) {
+	resp, _, err := a.WorkerControllerInstanceClient.DescribeWorkerControllerInstance(ctx, a.namespace, &deploymentpb.WorkerDeploymentVersion{DeploymentName: version.GetDeploymentName(), BuildId: version.GetBuildId()})
+	if err != nil {
+		if common.IsNotFoundError(err) {
+			return nil, nil
+		}
+
+		return nil, err
+	}
+
+	return wciValidationStatusToComputeStatus(resp.ValidationStatus), nil
+}
+
 func (a *VersionActivities) UpdateWorkerControllerInstance(ctx context.Context, input *deploymentspb.UpdateWorkerControllerInstanceInput) (*computepb.ComputeConfigSummary, error) {
 	upserts := scalingGroupUpdatesToWCI(input.GetUpsertScalingGroups())
 	resp, err := a.WorkerControllerInstanceClient.UpdateWorkerControllerInstance(ctx, a.namespace, input.GetVersion(), nil, input.GetIdentity(), upserts, input.GetRemoveScalingGroups())

service/worker/workerdeployment/version_workflow.go

@@ -345,6 +345,12 @@ func (d *VersionWorkflowRunner) run(ctx workflow.Context) error {
 		}
 	}
 
+	// When creating a compute provider and version together, there is a race condition between the two coming up. Making sure to have pulled
+	// the latest state from the compute provider if this happens to be the slower one.
+	if err := d.syncVersionDataToComputeStatus(ctx); err != nil {
+		return err
+	}
+
 	// Listen to signals in a different goroutine to make business logic clearer
 	workflow.Go(ctx, d.listenToSignals)
 
@@ -1244,6 +1250,27 @@ func (d *VersionWorkflowRunner) syncVersionStatusAfterDrainageStatusChange(ctx w
 	return d.syncVersionDataToTaskQueues(ctx, versionData)
 }
 
+// syncVersionDataToComputeStatus is a helper that syncs the compute status from WCI to the worker deployment version
+func (d *VersionWorkflowRunner) syncVersionDataToComputeStatus(ctx workflow.Context) error {
+	if workflow.GetVersion(ctx, "sync-compute-validation-status", workflow.DefaultVersion, 0) == workflow.DefaultVersion {
+		return nil
+	}
+
+	state := d.GetVersionState()
+
+	var result deploymentpb.ComputeStatus
+	resp := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, defaultActivityOptions), d.a.DescribeWorkerControllerInstanceStatus, state.GetVersion())
+	if err := resp.Get(ctx, &result); err != nil {
+		return err
+	}
+	if result.ProviderValidation == nil {
+		return nil
+	}
+
+	state.ComputeStatus = &result
+	return nil
+}
+
 // syncVersionDataToTaskQueues is a helper that syncs the provided version data to all task queues.
 // This function does NOT acquire the workflow lock - the caller is responsible for that.
 func (d *VersionWorkflowRunner) syncVersionDataToTaskQueues(ctx workflow.Context, versionData *deploymentspb.DeploymentVersionData) error {

service/worker/workerdeployment/version_workflow_test.go

@@ -54,6 +54,11 @@ func (s *VersionWorkflowSuite) SetupTest() {
 	}
 	s.env.RegisterWorkflowWithOptions(versionWorkflow, workflow.RegisterOptions{Name: WorkerDeploymentVersionWorkflowType})
 
+	var a *VersionActivities
+	s.env.OnActivity(a.DescribeWorkerControllerInstanceStatus, mock.Anything, mock.Anything).
+		Return(&deploymentpb.ComputeStatus{}, nil).
+		Maybe()
+
 	// Initialize an empty ClientImpl to use its helper methods
 	s.workerDeploymentClient = &ClientImpl{}
 }