Merge pull request #122 from cschleiden/signal-other-workflows

Signal other workflows

Author: cschleiden

README.md

@@ -348,6 +348,16 @@ func Workflow(ctx workflow.Context) error {
 }
 ```
 
+#### Signaling workflows from within workflows
+
+```go
+func Workflow(ctx workflow.Context) error {
+	if err := workflow.SignalWorkflow(ctx, "sub-instance-id", "signal-name", "value"); err != nil {
+		// Handle error
+	}
+}
+```
+
 ### Executing side effects
 
 Sometimes scheduling an activity is too much overhead for a simple side effect. For those scenarios you can use `workflow.SideEffect`. You can pass a func which will be executed only once inline with its result being recorded in the history. Subsequent executions of the workflow will return the previously recorded result.

backend/mysql/mysql.go

@@ -510,22 +510,27 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 	}
 
 	// Insert new workflow events
-	groupedEvents := history.EventsByWorkflowInstance(workflowEvents)
+	groupedEvents := history.EventsByWorkflowInstanceID(workflowEvents)
 
-	for targetInstance, events := range groupedEvents {
-		for _, event := range events {
-			if event.Type == history.EventType_WorkflowExecutionStarted {
-				a := event.Attributes.(*history.ExecutionStartedAttributes)
+	for targetInstanceID, events := range groupedEvents {
+		for _, m := range events {
+			if m.HistoryEvent.Type == history.EventType_WorkflowExecutionStarted {
+				a := m.HistoryEvent.Attributes.(*history.ExecutionStartedAttributes)
 				// Create new instance
-				if err := createInstance(ctx, tx, &targetInstance, a.Metadata, true); err != nil {
+				if err := createInstance(ctx, tx, m.WorkflowInstance, a.Metadata, true); err != nil {
 					return err
 				}
 
 				break
 			}
 		}
 
-		if err := insertPendingEvents(ctx, tx, targetInstance.InstanceID, events); err != nil {
+		historyEvents := []history.Event{}
+		for _, m := range events {
+			historyEvents = append(historyEvents, m.HistoryEvent)
+		}
+
+		if err := insertPendingEvents(ctx, tx, targetInstanceID, historyEvents); err != nil {
 			return fmt.Errorf("inserting messages: %w", err)
 		}
 	}

backend/redis/workflow.go

@@ -201,29 +201,29 @@ func (rb *redisBackend) CompleteWorkflowTask(
 	}
 
 	// Send new workflow events to the respective streams
-	groupedEvents := history.EventsByWorkflowInstance(workflowEvents)
-	for targetInstance, events := range groupedEvents {
+	groupedEvents := history.EventsByWorkflowInstanceID(workflowEvents)
+	for targetInstanceID, events := range groupedEvents {
 		// Insert pending events for target instance
-		for _, event := range events {
-			event := event
+		for _, m := range events {
+			m := m
 
-			if event.Type == history.EventType_WorkflowExecutionStarted {
+			if m.HistoryEvent.Type == history.EventType_WorkflowExecutionStarted {
 				// Create new instance
-				a := event.Attributes.(*history.ExecutionStartedAttributes)
-				if err := createInstanceP(ctx, p, &targetInstance, a.Metadata, true); err != nil {
+				a := m.HistoryEvent.Attributes.(*history.ExecutionStartedAttributes)
+				if err := createInstanceP(ctx, p, m.WorkflowInstance, a.Metadata, true); err != nil {
 					return err
 				}
 			}
 
 			// Add pending event to stream
-			if err := addEventToStreamP(ctx, p, pendingEventsKey(targetInstance.InstanceID), &event); err != nil {
+			if err := addEventToStreamP(ctx, p, pendingEventsKey(targetInstanceID), &m.HistoryEvent); err != nil {
 				return err
 			}
 		}
 
 		// Try to queue workflow task
-		if targetInstance != *instance {
-			if err := rb.workflowQueue.Enqueue(ctx, p, targetInstance.InstanceID, nil); err != nil {
+		if targetInstanceID != instance.InstanceID {
+			if err := rb.workflowQueue.Enqueue(ctx, p, targetInstanceID, nil); err != nil {
 				return fmt.Errorf("enqueuing workflow task: %w", err)
 			}
 		}

backend/sqlite/sqlite.go

@@ -399,14 +399,14 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 	}
 
 	// Insert new workflow events
-	groupedEvents := history.EventsByWorkflowInstance(workflowEvents)
+	groupedEvents := history.EventsByWorkflowInstanceID(workflowEvents)
 
-	for targetInstance, events := range groupedEvents {
-		for _, event := range events {
-			if event.Type == history.EventType_WorkflowExecutionStarted {
-				a := event.Attributes.(*history.ExecutionStartedAttributes)
+	for targetInstanceID, events := range groupedEvents {
+		for _, m := range events {
+			if m.HistoryEvent.Type == history.EventType_WorkflowExecutionStarted {
+				a := m.HistoryEvent.Attributes.(*history.ExecutionStartedAttributes)
 				// Create new instance
-				if err := createInstance(ctx, tx, &targetInstance, a.Metadata, true); err != nil {
+				if err := createInstance(ctx, tx, m.WorkflowInstance, a.Metadata, true); err != nil {
 					return err
 				}
 
@@ -415,7 +415,11 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 		}
 
 		// Insert pending events for target instance
-		if err := insertPendingEvents(ctx, tx, targetInstance.InstanceID, events); err != nil {
+		historyEvents := []history.Event{}
+		for _, m := range events {
+			historyEvents = append(historyEvents, m.HistoryEvent)
+		}
+		if err := insertPendingEvents(ctx, tx, targetInstanceID, historyEvents); err != nil {
 			return fmt.Errorf("inserting messages: %w", err)
 		}
 	}

backend/test/e2e.go

@@ -288,6 +288,40 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 				require.Equal(t, backend.ErrInstanceNotFound, err)
 			},
 		},
+		{
+			name: "SubWorkflow_Signal",
+			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {
+				swf := func(ctx workflow.Context, i int) (int, error) {
+					workflow.NewSignalChannel[string](ctx, "signal").Receive(ctx)
+
+					return i * 2, nil
+				}
+				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)
+
+					f := workflow.CreateSubWorkflowInstance[int](ctx, workflow.SubWorkflowOptions{
+						InstanceID: id,
+					}, swf, 1)
+
+					if err := workflow.SignalWorkflow(ctx, id, "signal", "hello"); err != nil {
+						return 0, err
+					}
+
+					return f.Get(ctx)
+				}
+				register(t, ctx, w, []interface{}{wf, swf}, nil)
+
+				instance := runWorkflow(t, ctx, c, wf)
+
+				r, err := client.GetWorkflowResult[int](ctx, c, instance, time.Second*20)
+				require.NoError(t, err)
+				require.Equal(t, 2, r)
+			},
+		},
 		{
 			name: "Timer_CancelWorkflowInstance",
 			f: func(t *testing.T, ctx context.Context, c client.Client, w worker.Worker, b TestBackend) {

internal/command/signal.go

@@ -0,0 +1,72 @@
+package command
+
+import (
+	"github.com/benbjohnson/clock"
+	"github.com/cschleiden/go-workflows/internal/core"
+	"github.com/cschleiden/go-workflows/internal/history"
+	"github.com/cschleiden/go-workflows/internal/payload"
+)
+
+type SignalWorkflowCommand struct {
+	command
+
+	Instance *core.WorkflowInstance
+
+	Name string
+	Arg  payload.Payload
+}
+
+var _ Command = (*SignalWorkflowCommand)(nil)
+
+func NewSignalWorkflowCommand(
+	id int64, workflowInstanceID, name string, arg payload.Payload,
+) *SignalWorkflowCommand {
+	return &SignalWorkflowCommand{
+		command: command{
+			state: CommandState_Pending,
+			id:    id,
+		},
+
+		Instance: core.NewWorkflowInstance(workflowInstanceID, ""), // TODO: Do we need a special identifier for an empty execution id?
+
+		Name: name,
+		Arg:  arg,
+	}
+}
+
+func (*SignalWorkflowCommand) Type() string {
+	return "WorkflowSignal"
+}
+
+func (c *SignalWorkflowCommand) Commit(clock clock.Clock) *CommandResult {
+	c.commit()
+
+	return &CommandResult{
+		// Record signal requested
+		Events: []history.Event{
+			history.NewPendingEvent(
+				clock.Now(),
+				history.EventType_SignalWorkflow,
+				&history.SignalWorkflowAttributes{
+					Name: c.Name,
+					Arg:  c.Arg,
+				},
+				history.ScheduleEventID(c.id),
+			),
+		},
+		// Send event to workflow instance
+		WorkflowEvents: []history.WorkflowEvent{
+			{
+				WorkflowInstance: c.Instance,
+				HistoryEvent: history.NewPendingEvent(
+					clock.Now(),
+					history.EventType_SignalReceived,
+					&history.SignalReceivedAttributes{
+						Name: c.Name,
+						Arg:  c.Arg,
+					},
+				),
+			},
+		},
+	}
+}

internal/history/grouping.go

@@ -1,18 +1,16 @@
 package history
 
-import "github.com/cschleiden/go-workflows/internal/core"
-
-func EventsByWorkflowInstance(events []WorkflowEvent) map[core.WorkflowInstance][]Event {
-	groupedEvents := make(map[core.WorkflowInstance][]Event)
+func EventsByWorkflowInstanceID(events []WorkflowEvent) map[string][]WorkflowEvent {
+	groupedEvents := make(map[string][]WorkflowEvent)
 
 	for _, m := range events {
 		instance := *m.WorkflowInstance
 
-		if _, ok := groupedEvents[instance]; !ok {
-			groupedEvents[instance] = []Event{}
+		if _, ok := groupedEvents[instance.InstanceID]; !ok {
+			groupedEvents[instance.InstanceID] = []WorkflowEvent{}
 		}
 
-		groupedEvents[instance] = append(groupedEvents[instance], m.HistoryEvent)
+		groupedEvents[instance.InstanceID] = append(groupedEvents[instance.InstanceID], m)
 	}
 
 	return groupedEvents

internal/history/grouping_test.go

@@ -12,17 +12,19 @@ import (
 func TestGrouping_MultipleEventsSameInstance(t *testing.T) {
 	id := uuid.NewString()
 
-	r := EventsByWorkflowInstance([]WorkflowEvent{
+	r := EventsByWorkflowInstanceID([]WorkflowEvent{
 		{
 			WorkflowInstance: core.NewWorkflowInstance(id, "exid"),
 			HistoryEvent:     NewPendingEvent(time.Now(), EventType_SubWorkflowScheduled, &SubWorkflowScheduledAttributes{}),
 		},
 		{
-			WorkflowInstance: core.NewWorkflowInstance(id, "exid"),
-			HistoryEvent:     NewPendingEvent(time.Now(), EventType_SubWorkflowScheduled, &SubWorkflowScheduledAttributes{}),
+			WorkflowInstance: core.NewWorkflowInstance(id, ""),
+			HistoryEvent:     NewPendingEvent(time.Now(), EventType_SignalReceived, &SubWorkflowScheduledAttributes{}),
 		},
 	})
 
 	require.Len(t, r, 1)
-	require.Len(t, r[*core.NewWorkflowInstance(id, "exid")], 2)
+	require.Len(t, r[id], 2)
+	require.Equal(t, r[id][0].HistoryEvent.Type, EventType_SubWorkflowScheduled)
+	require.Equal(t, r[id][1].HistoryEvent.Type, EventType_SignalReceived)
 }

internal/history/history.go

@@ -12,29 +12,50 @@ type EventType uint
 const (
 	_ EventType = iota
 
+	// Workflow has started
 	EventType_WorkflowExecutionStarted
+	// Workflow has finished
 	EventType_WorkflowExecutionFinished
+	// Workflow has been terminated (not yet used)
 	EventType_WorkflowExecutionTerminated
+	// Workflow has been canceled
 	EventType_WorkflowExecutionCanceled
 
+	// Workflow task has been started. This event is added to the history every time a workflow task is
+	// picked up by the worker.
 	EventType_WorkflowTaskStarted
 
+	// SubWorkflow has been scheduled
 	EventType_SubWorkflowScheduled
+	// SubWorkflow cancellation has been requested
 	EventType_SubWorkflowCancellationRequested
+	// SubWorkflow has completed
 	EventType_SubWorkflowCompleted
+	// SubWorkflow has failed
 	EventType_SubWorkflowFailed
 
+	// Activity task has been scheduled
 	EventType_ActivityScheduled
+	// Activity task has been completed
 	EventType_ActivityCompleted
+	// Activity task has failed
 	EventType_ActivityFailed
 
+	// Timer has been scheduled
 	EventType_TimerScheduled
+	// Timer has fired. This is the event received by a workflow when a previously scheduled timer fires.
 	EventType_TimerFired
+	// Timer has been canceled.
 	EventType_TimerCanceled
 
+	// Workflow has received a signal
 	EventType_SignalReceived
 
+	// Recorded result of a side-efect
 	EventType_SideEffectResult
+
+	// Signal other workflow
+	EventType_SignalWorkflow
 )
 
 func (et EventType) String() string {
@@ -79,6 +100,10 @@ func (et EventType) String() string {
 
 	case EventType_SideEffectResult:
 		return "SideEffectResult"
+
+	case EventType_SignalWorkflow:
+		return "WorkflowSignalRequested"
+
 	default:
 		return "Unknown"
 	}

internal/history/serialization.go

@@ -73,6 +73,9 @@ 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")
 	}