Fix out of order events bug + add test

This commit:
- Modifies executor.go to be aware of whether or not it has seen a WorkflowExecutionStarted event while executing tasks for the workflow instance it is tied to. If it has seen this event previously, execution continues as normal. If it hasn't seen the event previously, it checks through the events in the current task. If one of those events is WorkflowExecutionStarted, it shifts that event to the front of the task.NewEvents slice, ensuring that WorkflowExecutionStarted is processed immediately after the WorkflowTaskStarted event. This in turn ensures that the Scheduler the workflow structcontains is never nil, preventing the nil pointer dereference error described in issue #143.
- Adds a test inside of executor_test.go to verify the reordering behavior.

Author: zaataylor

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 {