Merge pull request #43 from famarting/orchestration-idempotency-test-attemp1

Orchestration idempotency test attemp1

Author: JoshVanL

backend/orchestration.go

@@ -170,16 +170,13 @@ func (w *orchestratorProcessor) applyWorkItem(ctx context.Context, wi *Orchestra
 	// New events from the work item are appended to the orchestration state, with duplicates automatically
 	// filtered out. If all events are filtered out, return false so that the caller knows not to execute
 	// the orchestration logic for an empty set of events.
-	added := 0
 	for _, e := range wi.NewEvents {
 		if err := runtimestate.AddEvent(wi.State, e); err != nil {
 			if err == runtimestate.ErrDuplicateEvent {
 				w.logger.Warnf("%v: dropping duplicate event: %v", wi.InstanceID, e)
 			} else {
 				w.logger.Warnf("%v: dropping event: %v, %v", wi.InstanceID, e, err)
 			}
-		} else {
-			added++
 		}
 
 		// Special case logic for specific event types
@@ -194,8 +191,8 @@ func (w *orchestratorProcessor) applyWorkItem(ctx context.Context, wi *Orchestra
 		}
 	}
 
-	if added == 0 {
-		w.logger.Warnf("%v: all new events were dropped", wi.InstanceID)
+	if len(wi.State.NewEvents) == 0 {
+		w.logger.Warnf("%v: no new events to process", wi.InstanceID)
 		return ctx, span, false
 	}
 

tests/worker_test.go

@@ -74,6 +74,77 @@ func Test_TryProcessSingleOrchestrationWorkItem_BasicFlow(t *testing.T) {
 	}, 1*time.Second, 100*time.Millisecond)
 
 	worker.StopAndDrain()
+
+	t.Logf("state.NewEvents: %v", state.NewEvents)
+	require.Len(t, state.NewEvents, 2)
+	require.NotNil(t, wi.State.NewEvents[0].GetOrchestratorStarted())
+	require.NotNil(t, wi.State.NewEvents[1].GetExecutionStarted())
+}
+
+func Test_TryProcessSingleOrchestrationWorkItem_Idempotency(t *testing.T) {
+	workflowID := "test123"
+	wi := &backend.OrchestrationWorkItem{
+		InstanceID: api.InstanceID(workflowID),
+		NewEvents: []*protos.HistoryEvent{
+			{
+				EventId:   -1,
+				Timestamp: timestamppb.New(time.Now()),
+				EventType: &protos.HistoryEvent_ExecutionStarted{
+					ExecutionStarted: &protos.ExecutionStartedEvent{
+						Name: "MyOrch",
+						OrchestrationInstance: &protos.OrchestrationInstance{
+							InstanceId:  workflowID,
+							ExecutionId: wrapperspb.String(uuid.New().String()),
+						},
+					},
+				},
+			},
+		},
+		State: runtimestate.NewOrchestrationRuntimeState(workflowID, nil, []*protos.HistoryEvent{}),
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	completed := atomic.Bool{}
+	be := mocks.NewBackend(t)
+	ex := mocks.NewExecutor(t)
+
+	callNumber := 0
+	ex.EXPECT().ExecuteOrchestrator(anyContext, wi.InstanceID, wi.State.OldEvents, mock.Anything).RunAndReturn(func(ctx context.Context, iid api.InstanceID, oldEvents []*protos.HistoryEvent, newEvents []*protos.HistoryEvent) (*protos.OrchestratorResponse, error) {
+		callNumber++
+		logger.Debugf("execute orchestrator called %d times", callNumber)
+		if callNumber == 1 {
+			return nil, errors.New("dummy error")
+		}
+		return &protos.OrchestratorResponse{}, nil
+	}).Times(2)
+
+	be.EXPECT().NextOrchestrationWorkItem(anyContext).Return(wi, nil).Once()
+	be.EXPECT().AbandonOrchestrationWorkItem(anyContext, wi).Return(nil).Once()
+
+	be.EXPECT().NextOrchestrationWorkItem(anyContext).Return(wi, nil).Once()
+	be.EXPECT().CompleteOrchestrationWorkItem(anyContext, wi).RunAndReturn(func(ctx context.Context, owi *backend.OrchestrationWorkItem) error {
+		completed.Store(true)
+		return nil
+	}).Once()
+
+	be.EXPECT().NextOrchestrationWorkItem(anyContext).Return(nil, errors.New("")).Once().Run(func(mock.Arguments) {
+		cancel()
+	})
+
+	worker := backend.NewOrchestrationWorker(be, ex, logger, backend.WithMaxParallelism(1))
+	worker.Start(ctx)
+
+	require.Eventually(t, completed.Load, 2*time.Second, 10*time.Millisecond)
+
+	worker.StopAndDrain()
+
+	t.Logf("state.NewEvents: %v", wi.State.NewEvents)
+	require.Len(t, wi.State.NewEvents, 3)
+	require.NotNil(t, wi.State.NewEvents[0].GetOrchestratorStarted())
+	require.NotNil(t, wi.State.NewEvents[1].GetExecutionStarted())
+	require.NotNil(t, wi.State.NewEvents[2].GetOrchestratorStarted())
 }
 
 func Test_TryProcessSingleOrchestrationWorkItem_ExecutionStartedAndCompleted(t *testing.T) {
@@ -156,6 +227,12 @@ func Test_TryProcessSingleOrchestrationWorkItem_ExecutionStartedAndCompleted(t *
 	}, 1*time.Second, 100*time.Millisecond)
 
 	worker.StopAndDrain()
+
+	t.Logf("state.NewEvents: %v", state.NewEvents)
+	require.Len(t, state.NewEvents, 3)
+	require.NotNil(t, wi.State.NewEvents[0].GetOrchestratorStarted())
+	require.NotNil(t, wi.State.NewEvents[1].GetExecutionStarted())
+	require.NotNil(t, wi.State.NewEvents[2].GetExecutionCompleted())
 }
 
 func Test_TaskWorker(t *testing.T) {