Merge pull request #397 from cschleiden/remove-single-worker-mode

Remove single worker mode and auto registration

Author: cschleiden

.vscode/settings.json

@@ -1,6 +1,6 @@
 {
   // Some integration tests take quite long to run, increase the overall limit for now
-  "go.testFlags": ["-timeout", "120s", "-race", "-count", "1"], // , "-short"],
+  "go.testFlags": ["-timeout", "120s", "-race", "-count", "1" , "-short"],
   "files.exclude": {
     "**/.git": true,
     "**/.svn": true,

docs/source/index.html.md

@@ -141,16 +141,12 @@ With the exception of the in-memory backend, we do not have to start the workflo
 For simpler scenarios where you don't need the separation between client and worker, you can use the `WorkflowOrchestrator` which combines both:
 
 ```go
-// Define the workflow with activities that will be auto-registered
-func AutoRegisteredWorkflow(ctx workflow.Context, input string) (string, error) {
-	// When using WorkflowOrchestrator, activities are automatically registered
-	// You can directly use any activity function without registering it first
+func MyWorkflow(ctx workflow.Context, input string) (string, error) {
 	r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1).Get(ctx)
 	if err != nil {
 		return "", err
 	}
 
-	// Another activity that will be auto-registered
 	r2, err := workflow.ExecuteActivity[string](ctx, workflow.DefaultActivityOptions, FormatResult, input, r1).Get(ctx)
 	if err != nil {
 		return "", err
@@ -159,7 +155,6 @@ func AutoRegisteredWorkflow(ctx workflow.Context, input string) (string, error)
 	return r2, nil
 }
 
-// Activities are automatically registered when using WorkflowOrchestrator
 func Activity1(ctx context.Context) (int, error) {
 	return 42, nil
 }
@@ -176,8 +171,10 @@ func main() {
 	// Create orchestrator instead of separate client and worker
 	orchestrator := worker.NewWorkflowOrchestrator(b, nil)
 
-	// Workflows are automatically registered
-	// Activities defined with InlineActivity don't need registration
+	// Register workflows and activities explicitly
+	orchestrator.RegisterWorkflow(MyWorkflow)
+	orchestrator.RegisterActivity(Activity1)
+	orchestrator.RegisterActivity(FormatResult)
 
 	// Start the orchestrator
 	if err := orchestrator.Start(ctx); err != nil {
@@ -187,7 +184,7 @@ func main() {
 	// Create workflow instance
 	wf, err := orchestrator.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
 		InstanceID: uuid.NewString(),
-	}, AutoRegisteredWorkflow, "input-for-workflow")
+	}, MyWorkflow, "input-for-workflow")
 	if err != nil {
 		panic("could not start workflow")
 	}
@@ -202,4 +199,4 @@ func main() {
 }
 ```
 
-The `WorkflowOrchestrator` provides automatic registration of workflows when passing workflow functions directly to `CreateWorkflowInstance`. For activities, the orchestrator automatically registers any activities used in the workflows via `ExecuteActivity` without explicit registration. Similarly, any sub-workflows created with `CreateSubWorkflowInstance` are registered automatically. This approach offers a unified API for workflow creation and execution in a single component, simplifying the development experience for simpler scenarios.
+The `WorkflowOrchestrator` provides a unified API for workflow creation and execution in a single component, combining both client and worker functionality. You need to explicitly register workflows and activities before starting the orchestrator, ensuring proper registration before any workflow or activity tasks are processed.

internal/contextvalue/singleworkermode.go

@@ -27,7 +27,6 @@ type WorkflowWorkerOptions struct {
 	WorkflowExecutorCache     executor.Cache
 	WorkflowExecutorCacheSize int
 	WorkflowExecutorCacheTTL  time.Duration
-	SingleWorkerMode          bool
 }
 
 func NewWorkflowWorker(
@@ -40,22 +39,20 @@ func NewWorkflowWorker(
 	}
 
 	tw := &WorkflowTaskWorker{
-		backend:          b,
-		registry:         registry,
-		cache:            options.WorkflowExecutorCache,
-		logger:           b.Options().Logger,
-		singleWorkerMode: options.SingleWorkerMode,
+		backend:  b,
+		registry: registry,
+		cache:    options.WorkflowExecutorCache,
+		logger:   b.Options().Logger,
 	}
 
 	return NewWorker(b, tw, &options.WorkerOptions)
 }
 
 type WorkflowTaskWorker struct {
-	backend          backend.Backend
-	registry         *registry.Registry
-	cache            executor.Cache
-	logger           *slog.Logger
-	singleWorkerMode bool
+	backend  backend.Backend
+	registry *registry.Registry
+	cache    executor.Cache
+	logger   *slog.Logger
 }
 
 func (wtw *WorkflowTaskWorker) Start(ctx context.Context, queues []workflow.Queue) error {
@@ -184,7 +181,6 @@ func (wtw *WorkflowTaskWorker) getExecutor(ctx context.Context, t *backend.Workf
 			t.Metadata,
 			clock.New(),
 			wtw.backend.Options().MaxHistorySize,
-			wtw.singleWorkerMode,
 		)
 		if err != nil {
 			return nil, fmt.Errorf("creating workflow task executor: %w", err)

internal/worker/workflow.go

@@ -25,14 +25,18 @@ func main() {
 		nil,
 	)
 
+	// Register workflows and activities explicitly
+	orchestrator.RegisterWorkflow(SimpleWorkflow)
+	orchestrator.RegisterActivity(ProcessMessage)
+
 	if err := orchestrator.Start(ctx); err != nil {
 		panic("could not start orchestrator: " + err.Error())
 	}
 
 	// Create instance ID
 	instanceID := uuid.NewString()
 
-	// Create and run workflow using the orchestrator - no explicit registration needed
+	// Create and run workflow using the orchestrator
 	instance, err := orchestrator.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
 		InstanceID: instanceID,
 	}, SimpleWorkflow, "Hello from orchestrator!")
@@ -69,7 +73,6 @@ func SimpleWorkflow(ctx workflow.Context, message string) (string, error) {
 }
 
 // ProcessMessage is a simple activity that processes a message
-// When orchestrator mode is enabled, this will be automatically registered
 func ProcessMessage(ctx context.Context, message string) (string, error) {
 	return message + " (processed by activity)", nil
 }

samples/orchestrator/main.go

@@ -369,7 +369,6 @@ func (wt *workflowTester[TResult]) Execute(ctx context.Context, args ...any) {
 				tw.metadata,
 				wt.clock,
 				wt.options.MaxHistorySize,
-				wt.options.SingleWorkerMode,
 			)
 			if err != nil {
 				panic(fmt.Errorf("could not create workflow executor: %v", err))

samples/orchestrator/orchestrator

@@ -42,9 +42,9 @@ type Options struct {
 	WorkflowWorkerOptions
 	ActivityWorkerOptions
 
-	// SingleWorkerMode enables automatic registration of workflows and activities
-	// when they are used in workflows. This is useful for simple scenarios where
-	// you don't want to explicitly register each workflow and activity.
+	// SingleWorkerMode enables optimizations for scenarios where only a single worker
+	// is processing tasks. This should only be enabled when you have exactly one worker
+	// instance running.
 	SingleWorkerMode bool
 }
 

tester/tester.go

@@ -35,7 +35,7 @@ func New(backend backend.Backend, options *Options) *Worker {
 		options = &DefaultOptions
 	}
 
-	workflowWorker := newWorkflowWorker(backend, registry, &options.WorkflowWorkerOptions, options.SingleWorkerMode)
+	workflowWorker := newWorkflowWorker(backend, registry, &options.WorkflowWorkerOptions)
 	activityWorker := newActivityWorker(backend, registry, &options.ActivityWorkerOptions)
 
 	// Register internal activities
@@ -46,7 +46,7 @@ func New(backend backend.Backend, options *Options) *Worker {
 func NewWorkflowWorker(backend backend.Backend, options *WorkflowWorkerOptions) *Worker {
 	registry := registry.New()
 
-	return newWorker(backend, registry, []worker{newWorkflowWorker(backend, registry, options, false)})
+	return newWorker(backend, registry, []worker{newWorkflowWorker(backend, registry, options)})
 }
 
 // NewActivityWorker creates a worker that only processes activities.
@@ -94,7 +94,7 @@ func newActivityWorker(backend backend.Backend, registry *registry.Registry, opt
 	return activityWorker
 }
 
-func newWorkflowWorker(backend backend.Backend, registry *registry.Registry, options *WorkflowWorkerOptions, singleWorkerMode bool) worker {
+func newWorkflowWorker(backend backend.Backend, registry *registry.Registry, options *WorkflowWorkerOptions) worker {
 	if options == nil {
 		options = &DefaultOptions.WorkflowWorkerOptions
 	}
@@ -110,7 +110,6 @@ func newWorkflowWorker(backend backend.Backend, registry *registry.Registry, opt
 		WorkflowExecutorCache:     options.WorkflowExecutorCache,
 		WorkflowExecutorCacheSize: options.WorkflowExecutorCacheSize,
 		WorkflowExecutorCacheTTL:  options.WorkflowExecutorCacheTTL,
-		SingleWorkerMode:          singleWorkerMode,
 	})
 
 	return workflowWorker

worker/options.go

@@ -2,12 +2,10 @@ package worker
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/client"
-	"github.com/cschleiden/go-workflows/internal/fn"
 	"github.com/cschleiden/go-workflows/registry"
 	"github.com/cschleiden/go-workflows/workflow"
 )
@@ -26,9 +24,8 @@ func NewWorkflowOrchestrator(backend backend.Backend, options *Options) *Workflo
 		options = &DefaultOptions
 	}
 
-	// Enable SingleWorkerMode automatically for the orchestrator
+	// Create copy of options without auto-enabling SingleWorkerMode
 	orchestratorOptions := *options
-	orchestratorOptions.SingleWorkerMode = true
 
 	// Set default pollers to 1 for orchestrator mode (unless explicitly overridden)
 	if orchestratorOptions.WorkflowPollers == DefaultOptions.WorkflowPollers {
@@ -41,8 +38,8 @@ func NewWorkflowOrchestrator(backend backend.Backend, options *Options) *Workflo
 	// Create registry that will be shared between worker and orchestrator
 	reg := registry.New()
 
-	// Create a regular worker with the registry and SingleWorkerMode enabled
-	workflowWorker := newWorkflowWorker(backend, reg, &orchestratorOptions.WorkflowWorkerOptions, orchestratorOptions.SingleWorkerMode)
+	// Create a regular worker with the registry
+	workflowWorker := newWorkflowWorker(backend, reg, &orchestratorOptions.WorkflowWorkerOptions)
 	activityWorker := newActivityWorker(backend, reg, &orchestratorOptions.ActivityWorkerOptions)
 	w := newWorker(backend, reg, []worker{workflowWorker, activityWorker})
 	c := client.New(backend)
@@ -57,7 +54,7 @@ func NewWorkflowOrchestrator(backend backend.Backend, options *Options) *Workflo
 	return orchestrator
 }
 
-// Start starts the worker with single worker mode enabled.
+// Start starts the worker.
 func (o *WorkflowOrchestrator) Start(ctx context.Context) error {
 	return o.worker.Start(ctx)
 }
@@ -68,21 +65,7 @@ func (o *WorkflowOrchestrator) WaitForCompletion() error {
 }
 
 // CreateWorkflowInstance creates a new workflow instance using the client.
-// Automatically registers the workflow if it's not already registered.
 func (o *WorkflowOrchestrator) CreateWorkflowInstance(ctx context.Context, options client.WorkflowInstanceOptions, wf workflow.Workflow, args ...any) (*workflow.Instance, error) {
-	// Check if the workflow is a function (not a string name) and register it if needed
-	if _, ok := wf.(string); !ok {
-		// It's a function reference, try to register it if not already registered
-		name := fn.Name(wf)
-		_, err := o.registry.GetWorkflow(name)
-		if err != nil {
-			// Workflow not found in registry, register it directly
-			if err := o.worker.RegisterWorkflow(wf); err != nil {
-				return nil, fmt.Errorf("auto-registering workflow %s: %w", name, err)
-			}
-		}
-	}
-
 	return o.Client.CreateWorkflowInstance(ctx, options, wf, args...)
 }
 
@@ -99,6 +82,16 @@ func (o *WorkflowOrchestrator) SignalWorkflow(ctx context.Context, instanceID st
 	return o.Client.SignalWorkflow(ctx, instanceID, name, arg)
 }
 
+// RegisterWorkflow registers a workflow with the orchestrator's registry.
+func (o *WorkflowOrchestrator) RegisterWorkflow(wf workflow.Workflow, opts ...registry.RegisterOption) error {
+	return o.worker.RegisterWorkflow(wf, opts...)
+}
+
+// RegisterActivity registers an activity with the orchestrator's registry.
+func (o *WorkflowOrchestrator) RegisterActivity(a workflow.Activity, opts ...registry.RegisterOption) error {
+	return o.worker.RegisterActivity(a, opts...)
+}
+
 // RemoveWorkflowInstance removes a workflow instance.
 func (o *WorkflowOrchestrator) RemoveWorkflowInstance(ctx context.Context, instance *workflow.Instance) error {
 	return o.Client.RemoveWorkflowInstance(ctx, instance)