[#784]: feature: dynamic (catch-all) workflows

.github/workflows/codeql-analysis.yml

@@ -30,7 +30,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
         with:
           # We must fetch at least the immediate parents so that if this is
           # a pull request then we can checkout the head.
@@ -45,7 +45,7 @@ jobs:
 
       # Initializes the CodeQL tools for scanning.
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@54f647b7e1bb85c95cddabcd46b0c578ec92bc1a # v4.36.3
+        uses: github/codeql-action/init@5595ccaf912efad79be6eef63a5619ff05969be3 # v4.37.6
         with:
           languages: ${{ matrix.language }}
           # If you wish to specify custom queries, you can do so here or in a config file.
@@ -56,7 +56,7 @@ jobs:
       # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
       # If this step fails, then you should remove it and run the build manually (see below)
       - name: Autobuild
-        uses: github/codeql-action/autobuild@8aad20d150bbac5944a9f9d289da16a4b0d87c1e # v4.36.2
+        uses: github/codeql-action/autobuild@5595ccaf912efad79be6eef63a5619ff05969be3 # v4.37.6
       # 📚 https://git.io/JvXDl
 
       # ✏️ If the Autobuild fails above, remove it and uncomment the following three lines
@@ -68,4 +68,4 @@ jobs:
       #   make release
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@8aad20d150bbac5944a9f9d289da16a4b0d87c1e # v4.36.2
+        uses: github/codeql-action/analyze@5595ccaf912efad79be6eef63a5619ff05969be3 # v4.37.6

.github/workflows/linters.yml

@@ -8,7 +8,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Check out code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
 
       - name: Set up Go
         uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0

.github/workflows/linux.yml

@@ -36,7 +36,7 @@ jobs:
           extensions: dom, sockets, grpc, curl, protobuf
 
       - name: Check out code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
         with:
           submodules: true
 
@@ -109,7 +109,7 @@ jobs:
           extensions: sockets
 
       - name: Check out code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
 
       - name: Get Composer Cache Directory
         id: composer-cache
@@ -178,7 +178,7 @@ jobs:
           extensions: sockets
 
       - name: Check out code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
 
       - name: Get Composer Cache Directory
         id: composer-cache
@@ -271,7 +271,7 @@ jobs:
           extensions: sockets
 
       - name: Check out code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
 
       - name: Get Composer Cache Directory
         id: composer-cache

aggregatedpool/workers.go

@@ -125,59 +125,85 @@ func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo
 	workers := make([]worker.Worker, 0, len(wi))
 
 	for i := range wi {
-		log.Debug("worker info", zap.Any("worker_info", wi[i]))
+		workerInfo := wi[i]
+		log.Debug("worker info", zap.Any("worker_info", workerInfo))
 
 		// Override to 0: RoadRunner manages worker lifecycle independently
-		wi[i].Options.WorkerStopTimeout = 0
+		workerInfo.Options.WorkerStopTimeout = 0
 
-		if wi[i].TaskQueue == "" {
-			wi[i].TaskQueue = temporalClient.DefaultNamespace
+		if workerInfo.TaskQueue == "" {
+			workerInfo.TaskQueue = temporalClient.DefaultNamespace
 		}
 
-		if wi[i].Options.Identity == "" {
-			wi[i].Options.Identity = fmt.Sprintf(
+		if workerInfo.Options.Identity == "" {
+			workerInfo.Options.Identity = fmt.Sprintf(
 				"roadrunner:%s:%s",
-				wi[i].TaskQueue,
+				workerInfo.TaskQueue,
 				uuid.NewString(),
 			)
 		}
 
-		wi[i].Options.Interceptors = append(wi[i].Options.Interceptors, resolved...)
-
-		wrk := worker.New(tc, wi[i].TaskQueue, wi[i].Options)
-
-		for j := 0; j < len(wi[i].Workflows); j++ {
-			wf := wi[i].Workflows[j]
+		workerInfo.Options.Interceptors = append(workerInfo.Options.Interceptors, resolved...)
+
+		wrk := worker.New(tc, workerInfo.TaskQueue, workerInfo.Options)
+		dynamicWorkflowRegistered := false
+
+		for _, wf := range workerInfo.Workflows {
+			// A dynamic workflow is the catch-all: register it via
+			// RegisterDynamicWorkflow (not by name) so it handles any workflow
+			// type that has no statically registered handler. The shared proxy
+			// (wDef) is a WorkflowDefinitionFactory, which RegisterDynamicWorkflow
+			// accepts; it forwards the real workflow type name to PHP.
+			if wf.Dynamic {
+				if dynamicWorkflowRegistered {
+					return nil, errors.E(
+						errors.Op("temporal_workers"),
+						errors.Errorf("multiple dynamic workflows configured on task queue %q", workerInfo.TaskQueue),
+					)
+				}
+
+				err := registerWorkflow(func() {
+					wrk.RegisterDynamicWorkflow(wDef, workflow.DynamicRegisterOptions{})
+				}, wf.Name, workerInfo.TaskQueue)
+				if err != nil {
+					return nil, err
+				}
+				dynamicWorkflowRegistered = true
+
+				log.Debug("dynamic workflow registered", zap.String(tq, workerInfo.TaskQueue), zap.Any("workflow name", wf.Name))
+
+				continue
+			}
 
 			err := registerWorkflow(func() {
 				wrk.RegisterWorkflowWithOptions(wDef, workflow.RegisterOptions{
 					Name:                          wf.Name,
 					VersioningBehavior:            wf.VersioningBehavior,
 					DisableAlreadyRegisteredCheck: false,
 				})
-			}, wf.Name, wi[i].TaskQueue)
+			}, wf.Name, workerInfo.TaskQueue)
 			if err != nil {
 				return nil, err
 			}
 
-			log.Debug("workflow registered", zap.String(tq, wi[i].TaskQueue), zap.Any("workflow name", wf.Name), zap.Int("versioning_behavior", int(wf.VersioningBehavior)))
+			log.Debug("workflow registered", zap.String(tq, workerInfo.TaskQueue), zap.Any("workflow name", wf.Name), zap.Int("versioning_behavior", int(wf.VersioningBehavior)))
 		}
 
 		if actDef.disableActivityWorkers {
-			log.Debug("activity workers disabled", zap.String(tq, wi[i].TaskQueue))
+			log.Debug("activity workers disabled", zap.String(tq, workerInfo.TaskQueue))
 			// add worker to the pool without activities
 			workers = append(workers, wrk)
 			continue
 		}
 
-		for j := 0; j < len(wi[i].Activities); j++ {
+		for _, activity := range workerInfo.Activities {
 			wrk.RegisterActivityWithOptions(actDef.execute, tActivity.RegisterOptions{
-				Name:                          wi[i].Activities[j].Name,
+				Name:                          activity.Name,
 				DisableAlreadyRegisteredCheck: false,
 				SkipInvalidStructFunctions:    false,
 			})
 
-			log.Debug("activity registered", zap.String(tq, wi[i].TaskQueue), zap.Any("workflow name", wi[i].Activities[j].Name))
+			log.Debug("activity registered", zap.String(tq, workerInfo.TaskQueue), zap.Any("workflow name", activity.Name))
 		}
 		// add worker to the pool
 		workers = append(workers, wrk)

aggregatedpool/workers_test.go

@@ -6,9 +6,12 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/temporalio/roadrunner-temporal/v5/api"
+	"github.com/temporalio/roadrunner-temporal/v5/internal"
 	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/converter"
 	sdkinterceptor "go.temporal.io/sdk/interceptor"
+	"go.uber.org/zap"
 )
 
 // mockPayloadConverter implements converter.PayloadConverter for testing.
@@ -278,6 +281,25 @@ func TestRegisterWorkflow_NonStringPanic_Handled(t *testing.T) {
 	assert.Contains(t, err.Error(), "42", "should preserve a non-string panic value")
 }
 
+func TestTemporalWorkers_MultipleDynamicWorkflows_ReturnsError(t *testing.T) {
+	temporalClient, err := client.NewLazyClient(client.Options{})
+	require.NoError(t, err)
+	t.Cleanup(temporalClient.Close)
+
+	workers := []*internal.WorkerInfo{{
+		TaskQueue: "default",
+		Workflows: []internal.WorkflowInfo{
+			{Name: "DynamicOne", Dynamic: true},
+			{Name: "DynamicTwo", Dynamic: true},
+		},
+	}}
+
+	_, err = TemporalWorkers(nil, nil, workers, zap.NewNop(), temporalClient, nil, nil)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "multiple dynamic workflows")
+	assert.Contains(t, err.Error(), "default")
+}
+
 func TestResolveDataConverters_EmptyMap_WithConfig(t *testing.T) {
 	_, err := ResolveDataConverters(map[string]converter.PayloadConverter{}, []string{"encoding/a"})
 	require.Error(t, err)

go.mod

@@ -1,6 +1,6 @@
 module github.com/temporalio/roadrunner-temporal/v5
 
-go 1.26.4
+go 1.26.5
 
 require (
 	github.com/goccy/go-json v0.10.6
@@ -13,10 +13,10 @@ require (
 	github.com/roadrunner-server/pool v1.1.3
 	github.com/stretchr/testify v1.11.1
 	github.com/uber-go/tally/v4 v4.1.17
-	go.temporal.io/api v1.62.14
-	go.temporal.io/sdk v1.44.1
+	go.temporal.io/api v1.63.4
+	go.temporal.io/sdk v1.47.0
 	go.temporal.io/sdk/contrib/tally v0.2.0
-	go.temporal.io/server v1.31.1
+	go.temporal.io/server v1.31.2
 	go.uber.org/zap v1.28.0
 	google.golang.org/protobuf v1.36.11
 )
@@ -30,18 +30,17 @@ 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/grpc-gateway/v2 v2.29.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nexus-rpc/sdk-go v0.6.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.2
+	github.com/prometheus/client_golang v1.24.1
 	github.com/prometheus/client_model v0.6.2 // indirect
-	github.com/prometheus/common v0.68.1 // indirect
-	github.com/prometheus/procfs v0.20.1 // indirect
+	github.com/prometheus/common v0.70.1 // indirect
+	github.com/prometheus/procfs v0.21.1 // indirect
 	github.com/roadrunner-server/goridge/v3 v3.8.3
 	github.com/robfig/cron v1.2.0 // indirect
-	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
 	github.com/stretchr/objx v0.5.3 // indirect
 	github.com/tklauser/go-sysconf v0.4.0 // indirect
@@ -50,19 +49,15 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
-	golang.org/x/sys v0.45.0 // indirect
-	golang.org/x/text v0.37.0 // indirect
+	golang.org/x/net v0.57.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
+	golang.org/x/text v0.40.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
-	google.golang.org/grpc v1.81.1
+	google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect
+	google.golang.org/grpc v1.83.0
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
-require (
-	github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect
-	go.opentelemetry.io/otel v1.44.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.44.0 // indirect
-)
+require github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect