Deliver signals to sub-workflows as activities

Author: cschleiden

backend/test/e2e.go

@@ -331,7 +331,7 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 						InstanceID: id,
 					}, swf, 1)
 
-					if err := workflow.SignalWorkflow(ctx, id, "signal", "hello"); err != nil {
+					if _, err := workflow.SignalWorkflow(ctx, id, "signal", "hello").Get(ctx); err != nil {
 						return 0, err
 					}
 
@@ -346,6 +346,30 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 				require.Equal(t, 2, r)
 			},
 		},
+		{
+			name: "SubWorkflow_Signal_BeforeStarting",
+			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
+				wf := func(ctx workflow.Context) (int, error) {
+					id, _ := workflow.SideEffect(ctx, func(ctx workflow.Context) string {
+						id := uuid.New().String()
+						workflow.Logger(ctx).Warn("side effect", "id", id)
+						return id
+					}).Get(ctx)
+
+					if _, err := workflow.SignalWorkflow(ctx, id, "signal", "hello").Get(ctx); err != nil {
+						return 0, err
+					}
+
+					return 42, nil
+				}
+				register(t, ctx, w, []interface{}{wf}, nil)
+
+				instance := runWorkflow(t, ctx, c, wf)
+
+				_, err := client.GetWorkflowResult[int](ctx, c, instance, time.Second*20)
+				require.ErrorContains(t, err, backend.ErrInstanceNotFound.Error())
+			},
+		},
 		{
 			name: "Timer_CancelWorkflowInstance",
 			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {

internal/command/signal.go

@@ -53,9 +53,6 @@ const (
 
 	// Recorded result of a side-efect
 	EventType_SideEffectResult
-
-	// Signal other workflow
-	EventType_SignalWorkflow
 )
 
 func (et EventType) String() string {
@@ -101,9 +98,6 @@ func (et EventType) String() string {
 	case EventType_SideEffectResult:
 		return "SideEffectResult"
 
-	case EventType_SignalWorkflow:
-		return "WorkflowSignalRequested"
-
 	default:
 		return "Unknown"
 	}

internal/command/signal_test.go

@@ -73,9 +73,6 @@ func DeserializeAttributes(eventType EventType, attributes []byte) (attr interfa
 	case EventType_SubWorkflowFailed:
 		attr = &SubWorkflowFailedAttributes{}
 
-	case EventType_SignalWorkflow:
-		attr = &SignalWorkflowAttributes{}
-
 	default:
 		return nil, errors.New("unknown event type when deserializing attributes")
 	}

internal/history/history.go

@@ -0,0 +1,17 @@
+package signals
+
+import (
+	"context"
+)
+
+type Signaler interface {
+	SignalWorkflow(ctx context.Context, instanceID string, name string, arg interface{}) error
+}
+
+type Activities struct {
+	Signaler Signaler
+}
+
+func (a *Activities) DeliverWorkflowSignal(ctx context.Context, instanceID, signalName string, arg interface{}) error {
+	return a.Signaler.SignalWorkflow(ctx, instanceID, signalName, arg)
+}

internal/history/serialization.go

@@ -320,9 +320,6 @@ func (e *executor) executeEvent(event history.Event) error {
 	case history.EventType_SubWorkflowCompleted:
 		err = e.handleSubWorkflowCompleted(event, event.Attributes.(*history.SubWorkflowCompletedAttributes))
 
-	case history.EventType_SignalWorkflow:
-		err = e.handleSignalWorkflow(event, event.Attributes.(*history.SignalWorkflowAttributes))
-
 	default:
 		return fmt.Errorf("unknown event type: %v", event.Type)
 	}
@@ -554,12 +551,10 @@ func (e *executor) handleSubWorkflowFailed(event history.Event, a *history.SubWo
 
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {
-		// TODO: Adjust
 		return fmt.Errorf("previous workflow execution scheduled a sub-workflow execution")
 	}
 
 	if _, ok := c.(*command.ScheduleSubWorkflowCommand); !ok {
-		// TODO: Adjust
 		return fmt.Errorf("previous workflow execution cancelled a sub-workflow execution, not: %v", c.Type())
 	}
 
@@ -582,12 +577,10 @@ func (e *executor) handleSubWorkflowCompleted(event history.Event, a *history.Su
 
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {
-		// TODO: Adjust
 		return fmt.Errorf("previous workflow execution cancelled a sub-workflow execution")
 	}
 
 	if _, ok := c.(*command.ScheduleSubWorkflowCommand); !ok {
-		// TODO: Adjust
 		return fmt.Errorf("previous workflow execution cancelled a sub-workflow execution, not: %v", c.Type())
 	}
 
@@ -603,22 +596,6 @@ func (e *executor) handleSignalReceived(event history.Event, a *history.SignalRe
 	return e.workflow.Continue()
 }
 
-func (e *executor) handleSignalWorkflow(event history.Event, a *history.SignalWorkflowAttributes) error {
-	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
-	if c == nil {
-		return fmt.Errorf("previous workflow execution requested a signal")
-	}
-
-	sewc, ok := c.(*command.SignalWorkflowCommand)
-	if !ok {
-		return fmt.Errorf("previous workflow execution requested to signal a workflow, not: %v", c.Type())
-	}
-
-	sewc.Done()
-
-	return e.workflow.Continue()
-}
-
 func (e *executor) handleSideEffectResult(event history.Event, a *history.SideEffectResultAttributes) error {
 	c := e.workflowState.CommandByScheduleEventID(event.ScheduleEventID)
 	if c == nil {

internal/history/signal_workflow.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/logger"
 	"github.com/cschleiden/go-workflows/internal/payload"
+	"github.com/cschleiden/go-workflows/internal/signals"
 	"github.com/cschleiden/go-workflows/internal/task"
 	"github.com/cschleiden/go-workflows/internal/workflow"
 	"github.com/cschleiden/go-workflows/log"
@@ -66,7 +67,7 @@ type WorkflowTester[TResult any] interface {
 }
 
 type testTimer struct {
-	// At is the timer this timer is scheduled for. This will advance the mock clock
+	// At is the time this timer is scheduled for. This will advance the mock clock
 	// to this timestamp
 	At time.Time
 
@@ -184,6 +185,10 @@ func NewWorkflowTester[TResult any](wf interface{}, opts ...WorkflowTesterOption
 		tracer: tracer,
 	}
 
+	// Register internal activities
+	signalActivities := &signals.Activities{Signaler: &signaler[TResult]{wt}}
+	registry.RegisterActivity(signalActivities)
+
 	// Always register the workflow under test
 	if err := wt.registry.RegisterWorkflow(wf); err != nil {
 		panic(fmt.Sprintf("could not workflow under test: %v", err))
@@ -635,3 +640,13 @@ func getNextWorkflowTask(wfi *core.WorkflowInstance, history []history.Event, ne
 		NewEvents:        newEvents,
 	}
 }
+
+type signaler[T any] struct {
+	wt *workflowTester[T]
+}
+
+func (s *signaler[T]) SignalWorkflow(ctx context.Context, instanceID string, name string, arg interface{}) error {
+	return nil
+}
+
+var _ signals.Signaler = (*signaler[any])(nil)

internal/signals/activities.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/internal/sync"
 	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/stretchr/testify/mock"
@@ -234,3 +235,31 @@ func workflowSignal(ctx workflow.Context) (string, error) {
 
 	return val, nil
 }
+
+func Test_SignalSubWorkflowBeforeScheduling(t *testing.T) {
+	tester := NewWorkflowTester[string](workflowSubWorkFlowsAndSignals)
+
+	tester.Execute()
+
+	require.True(t, tester.WorkflowFinished())
+	wfR, wfErr := tester.WorkflowResult()
+	require.Empty(t, wfErr)
+	require.IsType(t, "", wfR)
+}
+
+func workflowSubWorkFlowsAndSignals(ctx workflow.Context) (string, error) {
+	_, err := workflow.SignalWorkflow(ctx, "subworkflow", "test", "").Get(ctx)
+	if err != backend.ErrInstanceNotFound {
+		return "", err
+	}
+
+	workflow.CreateSubWorkflowInstance[int](ctx, workflow.SubWorkflowOptions{
+		InstanceID: "subworkflow",
+	}, workflowSum, 1, 2).Get(ctx)
+
+	return "finished without errors!", nil
+}
+
+func workflowSum(ctx workflow.Context, valA, valB int) (int, error) {
+	return valA + valB, nil
+}