Fix event grouping when delivering multiple events to a workflow instance in one checkpoint

Author: cschleiden

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

@@ -298,7 +298,9 @@ func EndToEndBackendTest(t *testing.T, setup func() TestBackend, teardown func(b
 				}
 				wf := func(ctx workflow.Context) (int, error) {
 					id, _ := workflow.SideEffect(ctx, func(ctx workflow.Context) string {
-						return uuid.New().String()
+						id := uuid.New().String()
+						workflow.Logger(ctx).Warn("side effect", "id", id)
+						return id
 					}).Get(ctx)
 
 					f := workflow.CreateSubWorkflowInstance[int](ctx, workflow.SubWorkflowOptions{

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)
 }