Merge pull request #145 from zaataylor/zaataylor/fix-out-of-order-events

Fix out of order events bug + add test

Author: cschleiden

internal/workflow/executor.go

@@ -40,17 +40,18 @@ type WorkflowExecutor interface {
 }
 
 type executor struct {
-	registry          *Registry
-	historyProvider   WorkflowHistoryProvider
-	workflow          *workflow
-	workflowTracer    *workflowtracer.WorkflowTracer
-	workflowState     *workflowstate.WfState
-	workflowCtx       sync.Context
-	workflowCtxCancel sync.CancelFunc
-	clock             clock.Clock
-	logger            log.Logger
-	tracer            trace.Tracer
-	lastSequenceID    int64
+	registry           *Registry
+	historyProvider    WorkflowHistoryProvider
+	workflow           *workflow
+	workflowTracer     *workflowtracer.WorkflowTracer
+	workflowState      *workflowstate.WfState
+	workflowCtx        sync.Context
+	workflowCtxCancel  sync.CancelFunc
+	clock              clock.Clock
+	logger             log.Logger
+	tracer             trace.Tracer
+	lastSequenceID     int64
+	wfStartedEventSeen bool
 }
 
 func NewExecutor(logger log.Logger, tracer trace.Tracer, registry *Registry, historyProvider WorkflowHistoryProvider, instance *core.WorkflowInstance, clock clock.Clock) (WorkflowExecutor, error) {
@@ -69,15 +70,16 @@ func NewExecutor(logger log.Logger, tracer trace.Tracer, registry *Registry, his
 	)
 
 	return &executor{
-		registry:          registry,
-		historyProvider:   historyProvider,
-		workflowTracer:    wfTracer,
-		workflowState:     s,
-		workflowCtx:       wfCtx,
-		workflowCtxCancel: cancel,
-		clock:             clock,
-		logger:            logger,
-		tracer:            tracer,
+		registry:           registry,
+		historyProvider:    historyProvider,
+		workflowTracer:     wfTracer,
+		workflowState:      s,
+		workflowCtx:        wfCtx,
+		workflowCtxCancel:  cancel,
+		clock:              clock,
+		logger:             logger,
+		tracer:             tracer,
+		wfStartedEventSeen: false,
 	}, nil
 }
 
@@ -141,6 +143,36 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 		return nil, fmt.Errorf("task has older history than current state, cannot execute")
 	}
 
+	// Potentially reorder new events here, protecting against
+	// cases in which new events are received for a workflow instance
+	// before the scheduler for that workflow instance has been
+	// created. To reorder, we find the WorkflowExecutionStarted
+	// event, then move it to the first position in t.NewEvents.
+	// t.NewEvents is modified in-place.
+	// See: https://github.com/cschleiden/go-workflows/issues/143
+	if !e.wfStartedEventSeen {
+		for i, ev := range t.NewEvents {
+			if ev.Type == history.EventType_WorkflowExecutionStarted {
+				if i > 0 {
+					// Shift elements before the WorkflowExecutionStarted
+					// event 1 index right, making space at index 0 to reinsert
+					// the WorkflowExecutionStarted event. Shifting instead of
+					// re-slicing and calling append() twice, i.e.:
+					// 		t.NewEvents = append(t.NewEvents[0:i], t.NewEvents[i+1:]...)
+					//		t.NewEvents = append([]history.Event{ev}, t.NewEvents...)
+					// is faster and avoids the possibility of copying
+					// slices if t.NewEvents is large
+					for j := i; j >= 1; j-- {
+						t.NewEvents[j] = t.NewEvents[j-1]
+					}
+					t.NewEvents[0] = ev
+				}
+				e.wfStartedEventSeen = true
+				break
+			}
+		}
+	}
+
 	// Always add a WorkflowTaskStarted event before executing new tasks
 	toExecute := []history.Event{e.createNewEvent(history.EventType_WorkflowTaskStarted, &history.WorkflowTaskStartedAttributes{})}
 	executedEvents := toExecute
@@ -205,9 +237,9 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) (*Executio
 	}, nil
 }
 
-func (e *executor) replayHistory(history []history.Event) error {
+func (e *executor) replayHistory(h []history.Event) error {
 	e.workflowState.SetReplaying(true)
-	for _, event := range history {
+	for _, event := range h {
 		if event.SequenceID < e.lastSequenceID {
 			e.logger.Panic("history has older events than current state")
 		}
@@ -216,6 +248,14 @@ func (e *executor) replayHistory(history []history.Event) error {
 			return err
 		}
 
+		// If we need to replay history before continuing execution of
+		// a new task, the executor must know if WorkflowExecutionStarted
+		// was seen during replay so it can determine if events should
+		// be reordered before it starts executing events for the new task
+		if event.Type == history.EventType_WorkflowExecutionStarted {
+			e.wfStartedEventSeen = true
+		}
+
 		e.lastSequenceID = event.SequenceID
 	}
 

internal/workflow/executor_test.go

@@ -585,6 +585,57 @@ func Test_Executor(t *testing.T) {
 				require.True(t, e.workflow.Completed())
 			},
 		},
+		{
+			name: "Reorder events to protect against nil deref error",
+			f: func(t *testing.T, r *Registry, e *executor, i *core.WorkflowInstance, hp *testHistoryProvider) {
+				workflow := func(ctx wf.Context) error {
+					return nil
+				}
+
+				r.RegisterWorkflow(workflow)
+
+				// Input for signal
+				input, err := converter.DefaultConverter.To("<Insert signal arg here>")
+				if err != nil {
+					require.NoError(t, err)
+				}
+				// Create a SignalReceived event
+				signalReceivedEvent := history.NewPendingEvent(
+					time.Now(),
+					history.EventType_SignalReceived,
+					&history.SignalReceivedAttributes{
+						Name: fn.Name(workflow),
+						Arg:  input,
+					},
+				)
+
+				// Create a TimerFired event
+				timerFiredEvent := history.NewPendingEvent(
+					time.Now(),
+					history.EventType_TimerFired,
+					&history.TimerFiredAttributes{
+						At: time.Now().Add(time.Second),
+					},
+				)
+
+				task := startWorkflowTask("instanceID", workflow)
+				// Here, we reorder the history so that the WorkflowExecutionStarted
+				// event does not appear first. This simulates the condition that caused
+				// the bug previously. To do so, we insert two events in front of the
+				// WorkflowExecutionStarted event.
+				require.False(t, e.wfStartedEventSeen)
+				task.NewEvents = append([]history.Event{signalReceivedEvent, timerFiredEvent}, task.NewEvents...)
+				result, err := e.ExecuteTask(context.Background(), task)
+				require.True(t, e.wfStartedEventSeen)
+				require.NoError(t, err)
+				require.NoError(t, e.workflow.err)
+				require.Len(t, result.Executed, 5)
+				// verify the events were executed in the reordered order
+				require.Equal(t, history.EventType_WorkflowExecutionStarted, result.Executed[1].Type)
+				require.Equal(t, history.EventType_SignalReceived, result.Executed[2].Type)
+				require.Equal(t, history.EventType_TimerFired, result.Executed[3].Type)
+			},
+		},
 	}
 
 	for _, tt := range tests {