Merge pull request #151 from cschleiden/cschleiden/handle-signal-errors

Deliver signals from within workflows as activities

Author: cschleiden

README.md

@@ -352,7 +352,7 @@ func Workflow(ctx workflow.Context) error {
 
 ```go
 func Workflow(ctx workflow.Context) error {
-	if err := workflow.SignalWorkflow(ctx, "sub-instance-id", "signal-name", "value"); err != nil {
+	if _, err := workflow.SignalWorkflow(ctx, "sub-instance-id", "signal-name", "value").Get(ctx); err != nil {
 		// Handle error
 	}
 }

backend/backend.go

@@ -34,6 +34,8 @@ type Backend interface {
 	GetWorkflowInstanceHistory(ctx context.Context, instance *workflow.Instance, lastSequenceID *int64) ([]history.Event, error)
 
 	// SignalWorkflow signals a running workflow instance
+	//
+	// If the given instance does not exist, it will return an error
 	SignalWorkflow(ctx context.Context, instanceID string, event history.Event) error
 
 	// GetWorkflowInstance returns a pending workflow task or nil if there are no pending worflow executions

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

@@ -192,11 +192,8 @@ func (ww *WorkflowWorker) getExecutor(ctx context.Context, t *task.Workflow) (wo
 	}
 
 	if !ok {
-		executor, err = workflow.NewExecutor(
+		executor = workflow.NewExecutor(
 			ww.backend.Logger(), ww.backend.Tracer(), ww.registry, ww.backend, t.WorkflowInstance, clock.New())
-		if err != nil {
-			return nil, fmt.Errorf("creating workflow executor: %w", err)
-		}
 	}
 
 	// Cache executor instance for future continuation tasks, or refresh last access time