Worker Versioning

client/client.go

@@ -188,6 +188,22 @@ type (
 	// WorkerBuildIDVersionSets is the response for Client.GetWorkerBuildIdCompatibility.
 	WorkerBuildIDVersionSets = internal.WorkerBuildIDVersionSets
 
+	// BuildIDOpAddNewIDInNewDefaultSet is an operation for UpdateWorkerBuildIdCompatibilityOptions
+	// to add a new BuildID in a new default set.
+	BuildIDOpAddNewIDInNewDefaultSet = internal.BuildIDOpAddNewIDInNewDefaultSet
+
+	// BuildIDOpAddNewCompatibleVersion is an operation for UpdateWorkerBuildIdCompatibilityOptions
+	// to add a new BuildID to an existing compatible set.
+	BuildIDOpAddNewCompatibleVersion = internal.BuildIDOpAddNewCompatibleVersion
+
+	// BuildIDOpPromoteSet is an operation for UpdateWorkerBuildIdCompatibilityOptions to promote a
+	// set to be the default set by targeting an existing BuildID.
+	BuildIDOpPromoteSet = internal.BuildIDOpPromoteSet
+
+	// BuildIDOpPromoteIDWithinSet is an operation for UpdateWorkerBuildIdCompatibilityOptions to
+	// promote a BuildID within a set to be the default.
+	BuildIDOpPromoteIDWithinSet = internal.BuildIDOpPromoteIDWithinSet
+
 	// Client is the client for starting and getting information about a workflow executions as well as
 	// completing activities asynchronously.
 	Client interface {

contrib/datadog/go.mod

@@ -3,7 +3,7 @@ module go.temporal.io/sdk/contrib/datadog
 go 1.16
 
 require (
-	github.com/stretchr/testify v1.8.2
+	github.com/stretchr/testify v1.8.3
 	go.temporal.io/sdk v1.12.0
 	gopkg.in/DataDog/dd-trace-go.v1 v1.42.0
 )

contrib/opentelemetry/go.mod

@@ -3,7 +3,7 @@ module go.temporal.io/sdk/contrib/opentelemetry
 go 1.16
 
 require (
-	github.com/stretchr/testify v1.8.2
+	github.com/stretchr/testify v1.8.3
 	go.opentelemetry.io/otel v1.2.0
 	go.opentelemetry.io/otel/sdk v1.2.0
 	go.opentelemetry.io/otel/trace v1.2.0

contrib/opentracing/go.mod

@@ -4,7 +4,7 @@ go 1.16
 
 require (
 	github.com/opentracing/opentracing-go v1.2.0
-	github.com/stretchr/testify v1.8.2
+	github.com/stretchr/testify v1.8.3
 	go.temporal.io/sdk v1.12.0
 )
 

contrib/tally/go.mod

@@ -3,7 +3,7 @@ module go.temporal.io/sdk/contrib/tally
 go 1.16
 
 require (
-	github.com/stretchr/testify v1.8.2
+	github.com/stretchr/testify v1.8.3
 	github.com/uber-go/tally/v4 v4.1.1
 	go.temporal.io/sdk v1.12.0
 )

go.mod

@@ -11,10 +11,10 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/pborman/uuid v1.2.1
 	github.com/robfig/cron v1.2.0
-	github.com/stretchr/testify v1.8.2
-	go.temporal.io/api v1.19.1-0.20230322213042-07fb271d475b
+	github.com/stretchr/testify v1.8.3
+	go.temporal.io/api v1.19.1-0.20230524162623-0e1dbb54f8e4
 	go.uber.org/atomic v1.9.0
-	golang.org/x/time v0.1.0
-	google.golang.org/grpc v1.54.0
+	golang.org/x/time v0.3.0
+	google.golang.org/grpc v1.55.0
 	google.golang.org/protobuf v1.30.0
 )

internal/activity.go

@@ -145,6 +145,10 @@ type (
 		// activities directly from the workflow task back to this worker which is
 		// faster than non-eager which may be dispatched to a separate worker.
 		DisableEagerExecution bool
+
+		// VersioningIntent specifies whether this activity should run on a worker with a compatible
+		// build ID or not. See temporal.VersioningIntent.
+		VersioningIntent VersioningIntent
 	}
 
 	// LocalActivityOptions stores local activity specific parameters that will be stored inside of a context.
@@ -208,9 +212,11 @@ func GetWorkerStopChannel(ctx context.Context) <-chan struct{} {
 // RecordActivityHeartbeat sends heartbeat for the currently executing activity
 // If the activity is either canceled (or) workflow/activity doesn't exist then we would cancel
 // the context with error context.Canceled.
-//  TODO: we don't have a way to distinguish between the two cases when context is canceled because
-//  context doesn't support overriding value of ctx.Error.
-//  TODO: Implement automatic heartbeating with cancellation through ctx.
+//
+//	TODO: we don't have a way to distinguish between the two cases when context is canceled because
+//	context doesn't support overriding value of ctx.Error.
+//	TODO: Implement automatic heartbeating with cancellation through ctx.
+//
 // details - the details that you provided here can be seen in the workflow when it receives TimeoutError, you
 // can check error TimeoutType()/Details().
 func RecordActivityHeartbeat(ctx context.Context, details ...interface{}) {

internal/error.go

@@ -177,6 +177,10 @@ type (
 		// Deprecated: WorkflowExecutionTimeout is deprecated and is never set or
 		// used internally.
 		WorkflowExecutionTimeout time.Duration
+
+		// VersioningIntent specifies whether the continued workflow should run on a worker with a
+		// compatible build ID or not. See VersioningIntent.
+		VersioningIntent VersioningIntent
 	}
 
 	// UnknownExternalWorkflowExecutionError can be returned when external workflow doesn't exist
@@ -416,14 +420,14 @@ func IsCanceledError(err error) bool {
 // If the workflow main function returns this error then the current execution is ended and
 // the new execution with same workflow ID is started automatically with options
 // provided to this function.
-//  ctx - use context to override any options for the new workflow like run timeout, task timeout, task queue.
-//	  if not mentioned it would use the defaults that the current workflow is using.
-//        ctx := WithWorkflowRunTimeout(ctx, 30 * time.Minute)
-//        ctx := WithWorkflowTaskTimeout(ctx, 5 * time.Second)
-//	  ctx := WithWorkflowTaskQueue(ctx, "example-group")
-//  wfn - workflow function. for new execution it can be different from the currently running.
-//  args - arguments for the new workflow.
 //
+//	 ctx - use context to override any options for the new workflow like run timeout, task timeout, task queue.
+//		  if not mentioned it would use the defaults that the current workflow is using.
+//	       ctx := WithWorkflowRunTimeout(ctx, 30 * time.Minute)
+//	       ctx := WithWorkflowTaskTimeout(ctx, 5 * time.Second)
+//		  ctx := WithWorkflowTaskQueue(ctx, "example-group")
+//	 wfn - workflow function. for new execution it can be different from the currently running.
+//	 args - arguments for the new workflow.
 func NewContinueAsNewError(ctx Context, wfn interface{}, args ...interface{}) error {
 	i := getWorkflowOutboundInterceptor(ctx)
 	// Put header on context before executing
@@ -460,6 +464,7 @@ func (wc *workflowEnvironmentInterceptor) NewContinueAsNewError(
 		WorkflowExecutionTimeout: options.WorkflowExecutionTimeout,
 		WorkflowRunTimeout:       options.WorkflowRunTimeout,
 		WorkflowTaskTimeout:      options.WorkflowTaskTimeout,
+		VersioningIntent:         options.VersioningIntent,
 	}
 }
 

internal/internal_activity.go

@@ -70,6 +70,7 @@ type (
 		OriginalTaskQueueName  string
 		RetryPolicy            *commonpb.RetryPolicy
 		DisableEagerExecution  bool
+		VersioningIntent       VersioningIntent
 	}
 
 	// ExecuteLocalActivityOptions options for executing a local activity

internal/internal_event_handlers.go

@@ -542,6 +542,8 @@ func (wc *workflowEnvironmentImpl) ExecuteChildWorkflow(
 	if len(params.CronSchedule) > 0 {
 		attributes.CronSchedule = params.CronSchedule
 	}
+	attributes.UseCompatibleVersion = determineUseCompatibleFlagForCommand(
+		params.VersioningIntent, wc.workflowInfo.TaskQueueName, params.TaskQueueName)
 
 	command, err := wc.commandsHelper.startChildWorkflowExecution(attributes)
 	if _, ok := err.(*childWorkflowExistsWithId); ok {
@@ -640,6 +642,8 @@ func (wc *workflowEnvironmentImpl) ExecuteActivity(parameters ExecuteActivityPar
 	// false just before request by the eager activity executor if eager activity
 	// execution is otherwise disallowed
 	scheduleTaskAttr.RequestEagerExecution = !parameters.DisableEagerExecution
+	scheduleTaskAttr.UseCompatibleVersion = determineUseCompatibleFlagForCommand(
+		parameters.VersioningIntent, wc.workflowInfo.TaskQueueName, parameters.TaskQueueName)
 
 	command := wc.commandsHelper.scheduleActivityTask(scheduleID, scheduleTaskAttr)
 	command.setData(&scheduledActivity{

internal/internal_task_handlers.go

@@ -127,6 +127,7 @@ type (
 		logger                   log.Logger
 		identity                 string
 		workerBuildID            string
+		useBuildIDForVersioning  bool
 		enableLoggingInReplay    bool
 		registry                 *registry
 		laTunnel                 *localActivityTunnel
@@ -158,6 +159,7 @@ type (
 		namespace                        string
 		defaultHeartbeatThrottleInterval time.Duration
 		maxHeartbeatThrottleInterval     time.Duration
+		versionStamp                     *commonpb.WorkerVersionStamp
 	}
 
 	// history wrapper method to help information about events.
@@ -422,7 +424,8 @@ func newWorkflowTaskHandler(params workerExecutionParameters, ppMgr pressurePoin
 		ppMgr:                    ppMgr,
 		metricsHandler:           params.MetricsHandler,
 		identity:                 params.Identity,
-		workerBuildID:            params.WorkerBuildID,
+		workerBuildID:            params.getBuildID(),
+		useBuildIDForVersioning:  params.UseBuildIDForVersioning,
 		enableLoggingInReplay:    params.EnableLoggingInReplay,
 		registry:                 registry,
 		workflowPanicPolicy:      params.WorkflowPanicPolicy,
@@ -905,7 +908,7 @@ ProcessEvents:
 			break ProcessEvents
 		}
 		if binaryChecksum == "" {
-			w.workflowInfo.BinaryChecksum = w.wth.getBuildID()
+			w.workflowInfo.BinaryChecksum = w.wth.workerBuildID
 		} else {
 			w.workflowInfo.BinaryChecksum = binaryChecksum
 		}
@@ -1586,16 +1589,20 @@ func (wth *workflowTaskHandlerImpl) completeWorkflow(
 		// Continue as new error.
 		metricsHandler.Counter(metrics.WorkflowContinueAsNewCounter).Inc(1)
 		closeCommand = createNewCommand(enumspb.COMMAND_TYPE_CONTINUE_AS_NEW_WORKFLOW_EXECUTION)
+
+		useCompat := determineUseCompatibleFlagForCommand(
+			contErr.VersioningIntent, workflowContext.workflowInfo.TaskQueueName, contErr.TaskQueueName)
 		closeCommand.Attributes = &commandpb.Command_ContinueAsNewWorkflowExecutionCommandAttributes{ContinueAsNewWorkflowExecutionCommandAttributes: &commandpb.ContinueAsNewWorkflowExecutionCommandAttributes{
-			WorkflowType:        &commonpb.WorkflowType{Name: contErr.WorkflowType.Name},
-			Input:               contErr.Input,
-			TaskQueue:           &taskqueuepb.TaskQueue{Name: contErr.TaskQueueName, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
-			WorkflowRunTimeout:  &contErr.WorkflowRunTimeout,
-			WorkflowTaskTimeout: &contErr.WorkflowTaskTimeout,
-			Header:              contErr.Header,
-			Memo:                workflowContext.workflowInfo.Memo,
-			SearchAttributes:    workflowContext.workflowInfo.SearchAttributes,
-			RetryPolicy:         convertToPBRetryPolicy(workflowContext.workflowInfo.RetryPolicy),
+			WorkflowType:         &commonpb.WorkflowType{Name: contErr.WorkflowType.Name},
+			Input:                contErr.Input,
+			TaskQueue:            &taskqueuepb.TaskQueue{Name: contErr.TaskQueueName, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
+			WorkflowRunTimeout:   &contErr.WorkflowRunTimeout,
+			WorkflowTaskTimeout:  &contErr.WorkflowTaskTimeout,
+			Header:               contErr.Header,
+			Memo:                 workflowContext.workflowInfo.Memo,
+			SearchAttributes:     workflowContext.workflowInfo.SearchAttributes,
+			RetryPolicy:          convertToPBRetryPolicy(workflowContext.workflowInfo.RetryPolicy),
+			UseCompatibleVersion: useCompat,
 		}}
 	} else if workflowContext.err != nil {
 		// Workflow failures
@@ -1648,27 +1655,25 @@ func (wth *workflowTaskHandlerImpl) completeWorkflow(
 		langUsedFlags = append(langUsedFlags, uint32(flag))
 	}
 
-	builtRequest := &workflowservice.RespondWorkflowTaskCompletedRequest{
+	return &workflowservice.RespondWorkflowTaskCompletedRequest{
 		TaskToken:                  task.TaskToken,
 		Commands:                   commands,
 		Messages:                   messages,
 		Identity:                   wth.identity,
 		ReturnNewWorkflowTask:      true,
 		ForceCreateNewWorkflowTask: forceNewWorkflowTask,
-		BinaryChecksum:             wth.getBuildID(),
+		BinaryChecksum:             wth.workerBuildID,
 		QueryResults:               queryResults,
 		Namespace:                  wth.namespace,
 		MeteringMetadata:           &commonpb.MeteringMetadata{NonfirstLocalActivityExecutionAttempts: nonfirstLAAttempts},
 		SdkMetadata: &sdk.WorkflowTaskCompletedMetadata{
 			LangUsedFlags: langUsedFlags,
 		},
+		WorkerVersionStamp: &commonpb.WorkerVersionStamp{
+			BuildId:       wth.workerBuildID,
+			UseVersioning: wth.useBuildIDForVersioning,
+		},
 	}
-	if wth.workerBuildID != "" {
-		builtRequest.WorkerVersionStamp = &commonpb.WorkerVersionStamp{
-			BuildId: wth.workerBuildID,
-		}
-	}
-	return builtRequest
 }
 
 func (wth *workflowTaskHandlerImpl) executeAnyPressurePoints(event *historypb.HistoryEvent, isInReplay bool) error {
@@ -1687,13 +1692,6 @@ func (wth *workflowTaskHandlerImpl) executeAnyPressurePoints(event *historypb.Hi
 	return nil
 }
 
-func (wth *workflowTaskHandlerImpl) getBuildID() string {
-	if wth.workerBuildID != "" {
-		return wth.workerBuildID
-	}
-	return getBinaryChecksum()
-}
-
 func newActivityTaskHandler(
 	service workflowservice.WorkflowServiceClient,
 	params workerExecutionParameters,
@@ -1724,6 +1722,10 @@ func newActivityTaskHandlerWithCustomProvider(
 		namespace:                        params.Namespace,
 		defaultHeartbeatThrottleInterval: params.DefaultHeartbeatThrottleInterval,
 		maxHeartbeatThrottleInterval:     params.MaxHeartbeatThrottleInterval,
+		versionStamp: &commonpb.WorkerVersionStamp{
+			BuildId:       params.getBuildID(),
+			UseVersioning: params.UseBuildIDForVersioning,
+		},
 	}
 }
 
@@ -1932,7 +1934,7 @@ func (ath *activityTaskHandlerImpl) Execute(taskQueue string, t *workflowservice
 		metricsHandler.Counter(metrics.UnregisteredActivityInvocationCounter).Inc(1)
 		return convertActivityResultToRespondRequest(ath.identity, t.TaskToken, nil,
 			NewActivityNotRegisteredError(activityType, ath.getRegisteredActivityNames()),
-			ath.dataConverter, ath.failureConverter, ath.namespace, false), nil
+			ath.dataConverter, ath.failureConverter, ath.namespace, false, ath.versionStamp), nil
 	}
 
 	// panic handler
@@ -1950,7 +1952,7 @@ func (ath *activityTaskHandlerImpl) Execute(taskQueue string, t *workflowservice
 			metricsHandler.Counter(metrics.ActivityTaskErrorCounter).Inc(1)
 			panicErr := newPanicError(p, st)
 			result = convertActivityResultToRespondRequest(ath.identity, t.TaskToken, nil, panicErr,
-				ath.dataConverter, ath.failureConverter, ath.namespace, false)
+				ath.dataConverter, ath.failureConverter, ath.namespace, false, ath.versionStamp)
 		}
 	}()
 
@@ -1990,7 +1992,7 @@ func (ath *activityTaskHandlerImpl) Execute(taskQueue string, t *workflowservice
 		)
 	}
 	return convertActivityResultToRespondRequest(ath.identity, t.TaskToken, output, err,
-		ath.dataConverter, ath.failureConverter, ath.namespace, isActivityCancel), nil
+		ath.dataConverter, ath.failureConverter, ath.namespace, isActivityCancel, ath.versionStamp), nil
 }
 
 func (ath *activityTaskHandlerImpl) getActivity(name string) activity {