Dynamic workflows and activities (#1946)

* WIP

* Added test for registering multiple dynamic activities

* varags of EncodedValue doesn't work

* test almost working, extra {} for some reason around

* need .name to get rid of brackets

* activities work now too

* clean up

* Fix lock location, add dynamic activity options,

* workflow_testsuite dynamic workflow test, fix other test

* rename, cleanup

* set unused field to _

* forgot a rename spot

Author: yuandrew

activity/activity.go

@@ -17,6 +17,9 @@ type (
 
 	// RegisterOptions consists of options for registering an activity.
 	RegisterOptions = internal.RegisterActivityOptions
+
+	// DynamicRegisterOptions consists of options for registering a dynamic activity.
+	DynamicRegisterOptions = internal.DynamicRegisterActivityOptions
 )
 
 // ErrResultPending is returned from activity's implementation to indicate the activity is not completed when the

client/client.go

@@ -1403,8 +1403,8 @@ func NewValue(data *commonpb.Payloads) converter.EncodedValue {
 }
 
 // NewValues creates a new [converter.EncodedValues] which can be used to decode binary data returned by Temporal. For example:
-// User had Activity.RecordHeartbeat(ctx, "my-heartbeat", 123) and then got response from calling [client.Client.DescribeWorkflowExecution].
-// The response contains binary field PendingActivityInfo.HeartbeatDetails,
+// User has Activity.RecordHeartbeat(ctx, "my-heartbeat", 123) and then got a response from calling [client.Client.DescribeWorkflowExecution].
+// The response contains the binary field PendingActivityInfo.HeartbeatDetails,
 // which can be decoded by using:
 //
 //	var result1 string

contrib/tools/workflowcheck/workflow/testdata/src/go.temporal.io/sdk/worker/worker.go

@@ -17,9 +17,11 @@ type Registry interface {
 type WorkflowRegistry interface {
 	RegisterWorkflow(w interface{})
 	RegisterWorkflowWithOptions(w interface{}, options workflow.RegisterOptions)
+	RegisterDynamicWorkflow(w interface{}, options workflow.DynamicRegisterOptions)
 }
 
 type ActivityRegistry interface {
 	RegisterActivity(a interface{})
 	RegisterActivityWithOptions(a interface{}, options activity.RegisterOptions)
+	RegisterDynamicActivity(a interface{}, options activity.DynamicRegisterOptions)
 }

internal/internal_worker.go

@@ -535,6 +535,10 @@ type registry struct {
 	workflowVersioningBehaviorMap map[string]VersioningBehavior
 	activityFuncMap               map[string]activity
 	activityAliasMap              map[string]string
+	dynamicWorkflow               interface{}
+	dynamicWorkflowOptions        DynamicRegisterWorkflowOptions
+	dynamicActivity               activity
+	_                             DynamicRegisterActivityOptions
 	interceptors                  []WorkerInterceptor
 }
 
@@ -567,7 +571,7 @@ func (r *registry) RegisterWorkflowWithOptions(
 	}
 	// Validate that it is a function
 	fnType := reflect.TypeOf(wf)
-	if err := validateFnFormat(fnType, true); err != nil {
+	if err := validateFnFormat(fnType, true, false); err != nil {
 		panic(err)
 	}
 	fnName, _ := getFunctionName(wf)
@@ -597,6 +601,29 @@ func (r *registry) RegisterWorkflowWithOptions(
 	}
 }
 
+func (r *registry) RegisterDynamicWorkflow(wf interface{}, options DynamicRegisterWorkflowOptions) {
+	r.Lock()
+	defer r.Unlock()
+	// Support direct registration of WorkflowDefinition
+	factory, ok := wf.(WorkflowDefinitionFactory)
+	if ok {
+		r.dynamicWorkflow = factory
+		r.dynamicWorkflowOptions = options
+		return
+	}
+
+	// Validate that it is a function
+	fnType := reflect.TypeOf(wf)
+	if err := validateFnFormat(fnType, true, true); err != nil {
+		panic(err)
+	}
+	if r.dynamicWorkflow != nil {
+		panic("dynamic workflow already registered")
+	}
+	r.dynamicWorkflow = wf
+	r.dynamicWorkflowOptions = options
+}
+
 func (r *registry) RegisterActivity(af interface{}) {
 	r.RegisterActivityWithOptions(af, RegisterActivityOptions{})
 }
@@ -626,7 +653,7 @@ func (r *registry) RegisterActivityWithOptions(
 		}
 		return
 	}
-	if err := validateFnFormat(fnType, false); err != nil {
+	if err := validateFnFormat(fnType, false, false); err != nil {
 		panic(err)
 	}
 	fnName, _ := getFunctionName(af)
@@ -669,7 +696,7 @@ func (r *registry) registerActivityStructWithOptions(aStruct interface{}, option
 			continue
 		}
 		name := method.Name
-		if err := validateFnFormat(method.Type, false); err != nil {
+		if err := validateFnFormat(method.Type, false, false); err != nil {
 			if options.SkipInvalidStructFunctions {
 				continue
 			}
@@ -691,6 +718,26 @@ func (r *registry) registerActivityStructWithOptions(aStruct interface{}, option
 	return nil
 }
 
+func (r *registry) RegisterDynamicActivity(af interface{}, options DynamicRegisterActivityOptions) {
+	r.Lock()
+	defer r.Unlock()
+	// Support direct registration of activity
+	a, ok := af.(activity)
+	if ok {
+		r.dynamicActivity = a
+		return
+	}
+	// Validate that it is a function
+	fnType := reflect.TypeOf(af)
+	if err := validateFnFormat(fnType, false, true); err != nil {
+		panic(err)
+	}
+	if r.dynamicActivity != nil {
+		panic("dynamic activity already registered")
+	}
+	r.dynamicActivity = &activityExecutor{name: "", fn: af, dynamic: true}
+}
+
 func (r *registry) RegisterNexusService(service *nexus.Service) {
 	if service.Name == "" {
 		panic(fmt.Errorf("tried to register a service with no name"))
@@ -715,8 +762,14 @@ func (r *registry) getWorkflowAlias(fnName string) (string, bool) {
 func (r *registry) getWorkflowFn(fnName string) (interface{}, bool) {
 	r.Lock()
 	defer r.Unlock()
-	fn, ok := r.workflowFuncMap[fnName]
-	return fn, ok
+	if fn, ok := r.workflowFuncMap[fnName]; ok {
+		return fn, ok
+	}
+
+	if r.dynamicWorkflow != nil {
+		return "dynamic", true
+	}
+	return nil, false
 }
 
 func (r *registry) getRegisteredWorkflowTypes() []string {
@@ -745,8 +798,13 @@ func (r *registry) addActivityWithLock(fnName string, a activity) {
 func (r *registry) GetActivity(fnName string) (activity, bool) {
 	r.Lock()
 	defer r.Unlock()
-	a, ok := r.activityFuncMap[fnName]
-	return a, ok
+	if a, ok := r.activityFuncMap[fnName]; ok {
+		return a, ok
+	}
+	if r.dynamicActivity != nil {
+		return r.dynamicActivity, true
+	}
+	return nil, false
 }
 
 func (r *registry) getActivityNoLock(fnName string) (activity, bool) {
@@ -757,10 +815,17 @@ func (r *registry) getActivityNoLock(fnName string) (activity, bool) {
 func (r *registry) getRegisteredActivities() []activity {
 	r.Lock()
 	defer r.Unlock()
-	activities := make([]activity, 0, len(r.activityFuncMap))
+	numActivities := len(r.activityFuncMap)
+	if r.dynamicActivity != nil {
+		numActivities++
+	}
+	activities := make([]activity, 0, numActivities)
 	for _, a := range r.activityFuncMap {
 		activities = append(activities, a)
 	}
+	if r.dynamicActivity != nil {
+		activities = append(activities, r.dynamicActivity)
+	}
 	return activities
 }
 
@@ -788,7 +853,12 @@ func (r *registry) getWorkflowDefinition(wt WorkflowType) (WorkflowDefinition, e
 	if ok {
 		return wdf.NewWorkflowDefinition(), nil
 	}
-	executor := &workflowExecutor{workflowType: lookup, fn: wf, interceptors: r.interceptors}
+	var dynamic bool
+	if d, ok := wf.(string); ok && d == "dynamic" {
+		wf = r.dynamicWorkflow
+		dynamic = true
+	}
+	executor := &workflowExecutor{workflowType: lookup, fn: wf, interceptors: r.interceptors, dynamic: dynamic}
 	return newSyncWorkflowDefinition(executor), nil
 }
 
@@ -799,8 +869,16 @@ func (r *registry) getWorkflowVersioningBehavior(wt WorkflowType) (VersioningBeh
 	}
 	r.Lock()
 	defer r.Unlock()
-	behavior := r.workflowVersioningBehaviorMap[lookup]
-	return behavior, behavior != VersioningBehaviorUnspecified
+	if behavior, ok := r.workflowVersioningBehaviorMap[lookup]; ok {
+		return behavior, behavior != VersioningBehaviorUnspecified
+	}
+	if r.dynamicWorkflowOptions.LoadDynamicRuntimeOptions != nil {
+		config := LoadDynamicRuntimeOptionsDetails{WorkflowType: wt}
+		if behavior, err := r.dynamicWorkflowOptions.LoadDynamicRuntimeOptions(config); err == nil {
+			return behavior.VersioningBehavior, true
+		}
+	}
+	return VersioningBehaviorUnspecified, false
 }
 
 func (r *registry) getNexusService(service string) *nexus.Service {
@@ -820,7 +898,7 @@ func (r *registry) getRegisteredNexusServices() []*nexus.Service {
 }
 
 // Validate function parameters.
-func validateFnFormat(fnType reflect.Type, isWorkflow bool) error {
+func validateFnFormat(fnType reflect.Type, isWorkflow, isDynamic bool) error {
 	if fnType.Kind() != reflect.Func {
 		return fmt.Errorf("expected a func as input but was %s", fnType.Kind())
 	}
@@ -845,6 +923,17 @@ func validateFnFormat(fnType reflect.Type, isWorkflow bool) error {
 		}
 	}
 
+	if isDynamic {
+		if fnType.NumIn() != 2 {
+			return fmt.Errorf(
+				"expected function to have two arguments, first being workflow.Context and second being an EncodedValues type, found %d arguments", fnType.NumIn(),
+			)
+		}
+		if fnType.In(1) != reflect.TypeOf((*converter.EncodedValues)(nil)).Elem() {
+			return fmt.Errorf("expected function to EncodedValues as second argument, got %s", fnType.In(1).Elem())
+		}
+	}
+
 	// Return values
 	// We expect either
 	// 	<result>, error
@@ -888,17 +977,25 @@ type workflowExecutor struct {
 	workflowType string
 	fn           interface{}
 	interceptors []WorkerInterceptor
+	dynamic      bool
 }
 
 func (we *workflowExecutor) Execute(ctx Context, input *commonpb.Payloads) (*commonpb.Payloads, error) {
 	dataConverter := WithWorkflowContext(ctx, getWorkflowEnvOptions(ctx).DataConverter)
 	fnType := reflect.TypeOf(we.fn)
 
-	args, err := decodeArgsToRawValues(dataConverter, fnType, input)
-	if err != nil {
-		return nil, fmt.Errorf(
-			"unable to decode the workflow function input payload with error: %w, function name: %v",
-			err, we.workflowType)
+	var args []interface{}
+	var err error
+	if we.dynamic {
+		// Dynamic workflows take in a single EncodedValues, encode all data into single EncodedValues
+		args = []interface{}{newEncodedValues(input, dataConverter)}
+	} else {
+		args, err = decodeArgsToRawValues(dataConverter, fnType, input)
+		if err != nil {
+			return nil, fmt.Errorf(
+				"unable to decode the workflow function input payload with error: %w, function name: %v",
+				err, we.workflowType)
+		}
 	}
 
 	envInterceptor := getWorkflowEnvironmentInterceptor(ctx)
@@ -918,6 +1015,7 @@ type activityExecutor struct {
 	name             string
 	fn               interface{}
 	skipInterceptors bool
+	dynamic          bool
 }
 
 func (ae *activityExecutor) ActivityType() ActivityType {
@@ -932,11 +1030,18 @@ func (ae *activityExecutor) Execute(ctx context.Context, input *commonpb.Payload
 	fnType := reflect.TypeOf(ae.fn)
 	dataConverter := getDataConverterFromActivityCtx(ctx)
 
-	args, err := decodeArgsToRawValues(dataConverter, fnType, input)
-	if err != nil {
-		return nil, fmt.Errorf(
-			"unable to decode the activity function input payload with error: %w for function name: %v",
-			err, ae.name)
+	var args []interface{}
+	var err error
+	if ae.dynamic {
+		// Dynamic activities take in a single EncodedValues, encode all data into single EncodedValues
+		args = []interface{}{newEncodedValues(input, dataConverter)}
+	} else {
+		args, err = decodeArgsToRawValues(dataConverter, fnType, input)
+		if err != nil {
+			return nil, fmt.Errorf(
+				"unable to decode the activity function input payload with error: %w for function name: %v",
+				err, ae.name)
+		}
 	}
 
 	return ae.ExecuteWithActualArgs(ctx, args)
@@ -1044,6 +1149,19 @@ func (aw *AggregatedWorker) RegisterWorkflowWithOptions(w interface{}, options R
 	aw.registry.RegisterWorkflowWithOptions(w, options)
 }
 
+// RegisterDynamicWorkflow registers dynamic workflow implementation with the AggregatedWorker
+func (aw *AggregatedWorker) RegisterDynamicWorkflow(w interface{}, options DynamicRegisterWorkflowOptions) {
+	if aw.workflowWorker == nil {
+		panic("workflow worker disabled, cannot register workflow")
+	}
+	if options.LoadDynamicRuntimeOptions == nil && aw.executionParams.UseBuildIDForVersioning &&
+		(aw.executionParams.DeploymentSeriesName != "" || aw.executionParams.WorkerDeploymentVersion != "") &&
+		aw.executionParams.DefaultVersioningBehavior == VersioningBehaviorUnspecified {
+		panic("dynamic workflow does not have a versioning behavior")
+	}
+	aw.registry.RegisterDynamicWorkflow(w, options)
+}
+
 // RegisterActivity registers activity implementation with the AggregatedWorker
 func (aw *AggregatedWorker) RegisterActivity(a interface{}) {
 	aw.registry.RegisterActivity(a)
@@ -1054,6 +1172,12 @@ func (aw *AggregatedWorker) RegisterActivityWithOptions(a interface{}, options R
 	aw.registry.RegisterActivityWithOptions(a, options)
 }
 
+// RegisterDynamicActivity registers the dynamic activity function with options.
+// Registering activities via a structure is not supported for dynamic activities.
+func (aw *AggregatedWorker) RegisterDynamicActivity(a interface{}, options DynamicRegisterActivityOptions) {
+	aw.registry.RegisterDynamicActivity(a, options)
+}
+
 func (aw *AggregatedWorker) RegisterNexusService(service *nexus.Service) {
 	if aw.started.Load() {
 		panic(errors.New("cannot register Nexus services after worker start"))
@@ -1334,6 +1458,11 @@ func (aw *WorkflowReplayer) RegisterWorkflowWithOptions(w interface{}, options R
 	aw.registry.RegisterWorkflowWithOptions(w, options)
 }
 
+// RegisterDynamicWorkflow registers a dynamic workflow function to replay
+func (aw *WorkflowReplayer) RegisterDynamicWorkflow(w interface{}, options DynamicRegisterWorkflowOptions) {
+	aw.registry.RegisterDynamicWorkflow(w, options)
+}
+
 // ReplayWorkflowHistoryWithOptions executes a single workflow task for the given history.
 // Use for testing the backwards compatibility of code changes and troubleshooting workflows in a debugger.
 // The logger is an optional parameter. Defaults to the noop logger.

internal/internal_worker_test.go

@@ -2983,3 +2983,30 @@ func (s *internalWorkerTestSuite) TestReservedTemporalName() {
 	require.Error(s.T(), err)
 	require.Contains(s.T(), err.Error(), temporalPrefixError)
 }
+
+func (s *internalWorkerTestSuite) TestRegisterMultipleDynamicWorkflow() {
+	var suite WorkflowTestSuite
+	env := suite.NewTestWorkflowEnvironment()
+	workflowFn1 := func(ctx Context, values converter.EncodedValues) error { return nil }
+	workflowFn2 := func(ctx Context, values converter.EncodedValues) error { return nil }
+	env.RegisterDynamicWorkflow(workflowFn1, DynamicRegisterWorkflowOptions{})
+	err := runAndCatchPanic(func() {
+		env.RegisterDynamicWorkflow(workflowFn2, DynamicRegisterWorkflowOptions{})
+	})
+	require.Error(s.T(), err)
+	require.Contains(s.T(), err.Error(), "dynamic workflow already registered")
+
+	// activity
+	activityFn1 := func(ctx context.Context, values converter.EncodedValues) error {
+		return nil
+	}
+	activityFn2 := func(ctx context.Context, values converter.EncodedValues) error {
+		return nil
+	}
+	env.RegisterDynamicActivity(activityFn1, DynamicRegisterActivityOptions{})
+	err = runAndCatchPanic(func() {
+		env.RegisterDynamicActivity(activityFn2, DynamicRegisterActivityOptions{})
+	})
+	require.Error(s.T(), err)
+	require.Contains(s.T(), err.Error(), "dynamic activity already registered")
+}

internal/internal_workflow_client.go

@@ -1665,6 +1665,7 @@ func createStartWorkflowInput(
 		return nil, err
 	}
 	workflowType, err := getWorkflowFunctionName(registry, workflow)
+
 	if err != nil {
 		return nil, err
 	}