Check params when scheduling workflows or activities

Author: cschleiden

backend/options.go

@@ -24,8 +24,14 @@ type Options struct {
 
 	StickyTimeout time.Duration
 
+	// WorkflowLockTimeout determines how long a workflow task can be locked for. If the workflow task is not completed
+	// by that timeframe, it's considered abandoned and another worker might pick it up.
+	//
+	// For long running workflow tasks, combine this with heartbearts.
 	WorkflowLockTimeout time.Duration
 
+	// ActivityLockTimeout determines how long an activity task can be locked for. If the activity task is not completed
+	// by that timeframe, it's considered abandoned and another worker might pick it up
 	ActivityLockTimeout time.Duration
 }
 

client/client.go

@@ -55,6 +55,11 @@ func New(backend backend.Backend) Client {
 }
 
 func (c *client) CreateWorkflowInstance(ctx context.Context, options WorkflowInstanceOptions, wf workflow.Workflow, args ...interface{}) (*workflow.Instance, error) {
+	// Check arguments
+	if !fn.ParamsMatch(wf, 1, args...) {
+		return nil, errors.New("arguments do not match workflow parameters")
+	}
+
 	inputs, err := a.ArgsToInputs(c.backend.Converter(), args...)
 	if err != nil {
 		return nil, fmt.Errorf("converting arguments: %w", err)

client/client_test.go

@@ -12,11 +12,33 @@ import (
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/logger"
+	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
 
+func Test_Client_CreateWorkflowInstance_ParamMismatch(t *testing.T) {
+	wf := func(workflow.Context, int) (int, error) {
+		return 0, nil
+	}
+
+	ctx := context.Background()
+
+	b := &backend.MockBackend{}
+	c := &client{
+		backend: b,
+		clock:   clock.New(),
+	}
+
+	result, err := c.CreateWorkflowInstance(ctx, WorkflowInstanceOptions{
+		InstanceID: "id",
+	}, wf, "foo")
+	require.Zero(t, result)
+	require.EqualError(t, err, "arguments do not match workflow parameters")
+	b.AssertExpectations(t)
+}
+
 func Test_Client_GetWorkflowResultTimeout(t *testing.T) {
 	instance := core.NewWorkflowInstance(uuid.NewString(), "test")
 

internal/fn/fn.go

@@ -29,3 +29,22 @@ func ReturnTypeMatch[TResult any](fn interface{}) bool {
 	t := *new(TResult)
 	return fnType.Out(0) == reflect.TypeOf(t)
 }
+
+func ParamsMatch(fn interface{}, skip int, args ...interface{}) bool {
+	fnType := reflect.TypeOf(fn)
+	if fnType.Kind() != reflect.Func {
+		return false
+	}
+
+	if fnType.NumIn() != skip+len(args) {
+		return false
+	}
+
+	for i, arg := range args {
+		if fnType.In(skip+i) != reflect.TypeOf(arg) {
+			return false
+		}
+	}
+
+	return true
+}

internal/fn/fn_test.go

@@ -113,3 +113,69 @@ func TestReturnTypeMatch(t *testing.T) {
 		})
 	}
 }
+
+func intParam(int) {
+}
+
+func stringParam(string) {
+}
+
+func mixedParams(context.Context, int, string) {
+}
+
+func TestParamsMatch(t *testing.T) {
+	tests := []struct {
+		name string
+		fn   func() bool
+		want bool
+	}{
+		{
+			name: "int match",
+			fn: func() bool {
+				return ParamsMatch(intParam, 0, 42)
+			},
+			want: true,
+		},
+		{
+			name: "int mismatch",
+			fn: func() bool {
+				return ParamsMatch(intParam, 0, "")
+			},
+			want: false,
+		},
+		{
+			name: "string mismatch",
+			fn: func() bool {
+				return ParamsMatch(stringParam, 0, 42)
+			},
+			want: false,
+		},
+		{
+			name: "mixed params",
+			fn: func() bool {
+				return ParamsMatch(mixedParams, 1, 42, "")
+			},
+			want: true,
+		},
+		{
+			name: "mixed params - no skip",
+			fn: func() bool {
+				return ParamsMatch(mixedParams, 0, 42, "")
+			},
+			want: false,
+		},
+		{
+			name: "mixed params - wrong params",
+			fn: func() bool {
+				return ParamsMatch(mixedParams, 1, "", 42)
+			},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.fn()
+			require.Equal(t, tt.want, got)
+		})
+	}
+}

workflow/activity.go

@@ -38,11 +38,18 @@ func executeActivity[TResult any](ctx Context, options ActivityOptions, attempt
 		return f
 	}
 
+	// Check return type
 	if !fn.ReturnTypeMatch[TResult](activity) {
 		f.Set(*new(TResult), fmt.Errorf("activity return type does not match expected type"))
 		return f
 	}
 
+	// Check arguments
+	if !fn.ParamsMatch(activity, 1, args...) {
+		f.Set(*new(TResult), fmt.Errorf("activity arguments do not match expected types"))
+		return f
+	}
+
 	cv := converter.GetConverter(ctx)
 	inputs, err := a.ArgsToInputs(cv, args...)
 	if err != nil {

workflow/activity_test.go

@@ -14,7 +14,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
-func Test_executeActivity_ParamMismatch(t *testing.T) {
+func Test_executeActivity_ResultMismatch(t *testing.T) {
 	a := func(ctx Context) (int, error) {
 		return 42, nil
 	}
@@ -38,3 +38,27 @@ func Test_executeActivity_ParamMismatch(t *testing.T) {
 	c.Execute()
 	require.True(t, c.Finished())
 }
+func Test_executeActivity_ParamMismatch(t *testing.T) {
+	a := func(ctx Context, s string, n int) (int, error) {
+		return 42, nil
+	}
+
+	ctx := sync.Background()
+	ctx = converter.WithConverter(ctx, converter.DefaultConverter)
+	ctx = workflowstate.WithWorkflowState(
+		ctx,
+		workflowstate.NewWorkflowState(core.NewWorkflowInstance("a", ""), logger.NewDefaultLogger(), clock.New()),
+	)
+	ctx = workflowtracer.WithWorkflowTracer(ctx, workflowtracer.New(trace.NewNoopTracerProvider().Tracer("test")))
+
+	c := sync.NewCoroutine(ctx, func(ctx sync.Context) error {
+		f := executeActivity[int](ctx, DefaultActivityOptions, 1, a)
+		_, err := f.Get(ctx)
+		require.Error(t, err)
+
+		return nil
+	})
+
+	c.Execute()
+	require.True(t, c.Finished())
+}

workflow/subworkflow.go

@@ -48,11 +48,18 @@ func createSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflo
 		return f
 	}
 
+	// Check return type
 	if !fn.ReturnTypeMatch[TResult](wf) {
 		f.Set(*new(TResult), fmt.Errorf("subworkflow return type does not match expected type"))
 		return f
 	}
 
+	// Check arguments
+	if !fn.ParamsMatch(wf, 1, args...) {
+		f.Set(*new(TResult), fmt.Errorf("subworkflow arguments do not match expected types"))
+		return f
+	}
+
 	name := fn.Name(wf)
 
 	cv := converter.GetConverter(ctx)

workflow/subworkflow_test.go

@@ -15,6 +15,31 @@ import (
 )
 
 func Test_createSubWorkflowInstance_ParamMismatch(t *testing.T) {
+	wf := func(Context, int) (int, error) {
+		return 42, nil
+	}
+
+	ctx := sync.Background()
+	ctx = converter.WithConverter(ctx, converter.DefaultConverter)
+	ctx = workflowstate.WithWorkflowState(
+		ctx,
+		workflowstate.NewWorkflowState(core.NewWorkflowInstance("a", ""), logger.NewDefaultLogger(), clock.New()),
+	)
+	ctx = workflowtracer.WithWorkflowTracer(ctx, workflowtracer.New(trace.NewNoopTracerProvider().Tracer("test")))
+
+	c := sync.NewCoroutine(ctx, func(ctx sync.Context) error {
+		f := createSubWorkflowInstance[int](ctx, DefaultSubWorkflowOptions, 1, wf, "foo")
+		_, err := f.Get(ctx)
+		require.Error(t, err)
+
+		return nil
+	})
+
+	c.Execute()
+	require.True(t, c.Finished())
+}
+
+func Test_createSubWorkflowInstance_ReturnMismatch(t *testing.T) {
 	wf := func(ctx Context) (int, error) {
 		return 42, nil
 	}