[#680]feature: workflow versioning behavior

Author: rustatian

aggregatedpool/workers.go

@@ -18,37 +18,8 @@ const tq = "taskqueue"
 func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo, log *zap.Logger, tc temporalClient.Client, interceptors map[string]api.Interceptor) ([]worker.Worker, error) {
 	workers := make([]worker.Worker, 0, 1)
 
-	for i := 0; i < len(wi); i++ {
-		log.Debug("worker info",
-			zap.String(tq, wi[i].TaskQueue),
-			zap.Any("flags", wi[i].Flags),
-			zap.Int("num_workflows", len(wi[i].Workflows)),
-			zap.Int("num_activities", len(wi[i].Activities)),
-			zap.Int("max_concurrent_activity_execution_size", wi[i].Options.MaxConcurrentActivityExecutionSize),
-			zap.Float64("worker_activities_per_second", wi[i].Options.WorkerActivitiesPerSecond),
-			zap.Int("max_concurrent_local_activity_execution_size", wi[i].Options.MaxConcurrentLocalActivityExecutionSize),
-			zap.Float64("worker_local_activities_per_second", wi[i].Options.WorkerLocalActivitiesPerSecond),
-			zap.Float64("task_queue_activities_per_second", wi[i].Options.TaskQueueActivitiesPerSecond),
-			zap.Int("max_concurrent_activity_task_pollers", wi[i].Options.MaxConcurrentActivityTaskPollers),
-			zap.Int("max_concurrent_workflow_task_pollers", wi[i].Options.MaxConcurrentWorkflowTaskPollers),
-			zap.Bool("enable_logging_in_replay", wi[i].Options.EnableLoggingInReplay),
-			zap.String("workflow_panic_policy", policyToString(wi[i].Options.WorkflowPanicPolicy)),
-			zap.Duration("worker_stop_timeout", wi[i].Options.WorkerStopTimeout),
-			zap.Bool("enable_session_worker", wi[i].Options.EnableSessionWorker),
-			zap.Int("max_concurrent_session_execution_size", wi[i].Options.MaxConcurrentSessionExecutionSize),
-			zap.Bool("disable_workflow_worker", wi[i].Options.DisableWorkflowWorker),
-			zap.Bool("local_activity_worker_only", wi[i].Options.LocalActivityWorkerOnly),
-			zap.String("identity", wi[i].Options.Identity),
-			zap.Duration("deadlock_detection_timeout", wi[i].Options.DeadlockDetectionTimeout),
-			zap.Duration("max_heartbeat_throttle_interval", wi[i].Options.MaxHeartbeatThrottleInterval),
-			zap.Duration("default_heartbeat_throttle_interval", wi[i].Options.DefaultHeartbeatThrottleInterval),
-			zap.Bool("disable_eager_activities", wi[i].Options.DisableEagerActivities),
-			zap.Int("max_concurrent_eager_activity_execution_size", wi[i].Options.MaxConcurrentEagerActivityExecutionSize),
-			zap.Bool("disable_registration_aliasing", wi[i].Options.DisableRegistrationAliasing),
-			zap.Bool("use_versioning", wi[i].Options.DeploymentOptions.UseVersioning),
-			zap.String("build_id", wi[i].Options.DeploymentOptions.Version.BuildId),
-			zap.String("deployment_name", wi[i].Options.DeploymentOptions.Version.DeploymentName),
-		)
+	for i := range wi {
+		log.Debug("worker info", zap.Any("worker_info", wi[i]))
 
 		// just to be sure
 		wi[i].Options.WorkerStopTimeout = 0
@@ -59,7 +30,7 @@ func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo
 
 		if wi[i].Options.Identity == "" {
 			wi[i].Options.Identity = fmt.Sprintf(
-				"%s:%s",
+				"roadrunner:%s:%s",
 				wi[i].TaskQueue,
 				uuid.NewString(),
 			)
@@ -76,15 +47,18 @@ func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo
 		for j := 0; j < len(wi[i].Workflows); j++ {
 			wrk.RegisterWorkflowWithOptions(wDef, workflow.RegisterOptions{
 				Name:                          wi[i].Workflows[j].Name,
+				VersioningBehavior:            wi[i].Workflows[j].VersioningBehavior,
 				DisableAlreadyRegisteredCheck: false,
 			})
 
-			log.Debug("workflow registered", zap.String(tq, wi[i].TaskQueue), zap.Any("workflow name", wi[i].Workflows[j].Name))
+			log.Debug("workflow registered", zap.String(tq, wi[i].TaskQueue), zap.Any("workflow name", wi[i].Workflows[j].Name), zap.Int("versioning_behavior", int(wi[i].Workflows[j].VersioningBehavior)))
 		}
 
 		if actDef.disableActivityWorkers {
 			log.Debug("activity workers disabled", zap.String(tq, wi[i].TaskQueue))
-			goto RegisterWorkflows
+			// add worker to the pool without activities
+			workers = append(workers, wrk)
+			continue
 		}
 
 		for j := 0; j < len(wi[i].Activities); j++ {
@@ -96,32 +70,11 @@ func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo
 
 			log.Debug("activity registered", zap.String(tq, wi[i].TaskQueue), zap.Any("workflow name", wi[i].Activities[j].Name))
 		}
-
-	RegisterWorkflows:
-		for j := 0; j < len(wi[i].Workflows); j++ {
-			wrk.RegisterWorkflowWithOptions(wDef, workflow.RegisterOptions{
-				Name:                          wi[i].Workflows[j].Name,
-				DisableAlreadyRegisteredCheck: false,
-			})
-
-			log.Debug("workflow registered", zap.String(tq, wi[i].TaskQueue), zap.Any("workflow name", wi[i].Workflows[j].Name))
-		}
-
+		// add worker to the pool
 		workers = append(workers, wrk)
 	}
 
 	log.Debug("workers initialized", zap.Int("num_workers", len(workers)))
 
 	return workers, nil
 }
-
-func policyToString(enum worker.WorkflowPanicPolicy) string {
-	switch enum {
-	case worker.BlockWorkflow:
-		return "block"
-	case worker.FailWorkflow:
-		return "fail"
-	default:
-		return "unknown"
-	}
-}

go.mod

@@ -1,8 +1,6 @@
 module github.com/temporalio/roadrunner-temporal/v5
 
-go 1.24.5
-
-toolchain go1.24.6
+go 1.25
 
 require (
 	github.com/goccy/go-json v0.10.5
@@ -12,14 +10,22 @@ require (
 	github.com/roadrunner-server/errors v1.4.1
 	github.com/roadrunner-server/events v1.0.1
 	github.com/roadrunner-server/pool v1.1.3
-	github.com/stretchr/testify v1.11.0
+	github.com/stretchr/testify v1.11.1
 	github.com/uber-go/tally/v4 v4.1.17
 	go.temporal.io/api v1.52.0
-	go.temporal.io/sdk v1.35.0
+	go.temporal.io/sdk v1.36.0
 	go.temporal.io/sdk/contrib/tally v0.2.0
 	go.temporal.io/server v1.28.1
 	go.uber.org/zap v1.27.0
-	google.golang.org/protobuf v1.36.7
+	google.golang.org/protobuf v1.36.8
+)
+
+require (
+	github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc // indirect
+	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 // indirect
+	go.opentelemetry.io/otel v1.38.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.38.0 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 )
 
 replace github.com/uber-go/tally/v4 => github.com/uber-go/tally/v4 v4.1.10
@@ -33,15 +39,14 @@ require (
 	github.com/go-ole/go-ole v1.3.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/mock v1.7.0-rc.1 // indirect
-	github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nexus-rpc/sdk-go v0.4.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_golang v1.23.0
+	github.com/prometheus/client_golang v1.23.1
 	github.com/prometheus/client_model v0.6.2 // indirect
-	github.com/prometheus/common v0.65.0 // indirect
+	github.com/prometheus/common v0.66.0 // indirect
 	github.com/prometheus/procfs v0.17.0 // indirect
 	github.com/roadrunner-server/goridge/v3 v3.8.3
 	github.com/robfig/cron v1.2.0 // indirect
@@ -59,8 +64,8 @@ require (
 	golang.org/x/sys v0.35.0 // indirect
 	golang.org/x/text v0.28.0 // indirect
 	golang.org/x/time v0.12.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20250811230008-5f3141c8851a // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250811230008-5f3141c8851a // indirect
-	google.golang.org/grpc v1.74.2
+	google.golang.org/genproto/googleapis/api v0.0.0-20250826171959-ef028d996bc1 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250826171959-ef028d996bc1 // indirect
+	google.golang.org/grpc v1.75.0
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )