Add unit test for replay with side effect (#590)

temporalio · Oct 14, 2021 · f33117f · f33117f
1 parent e24f673
commit f33117f
Show file tree

Hide file tree

Showing 2 changed files with 164 additions and 0 deletions.
diff --git a/internal/internal_task_handlers_test.go b/internal/internal_task_handlers_test.go
@@ -241,6 +241,13 @@ func createTestEventWorkflowTaskFailed(eventID int64, attr *historypb.WorkflowTa
 		Attributes: &historypb.HistoryEvent_WorkflowTaskFailedEventAttributes{WorkflowTaskFailedEventAttributes: attr}}
 }
 
+func createTestEventWorkflowTaskTimedOut(eventID int64, attr *historypb.WorkflowTaskTimedOutEventAttributes) *historypb.HistoryEvent {
+	return &historypb.HistoryEvent{
+		EventId:    eventID,
+		EventType:  enumspb.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT,
+		Attributes: &historypb.HistoryEvent_WorkflowTaskTimedOutEventAttributes{WorkflowTaskTimedOutEventAttributes: attr}}
+}
+
 func createTestEventSignalExternalWorkflowExecutionFailed(eventID int64, attr *historypb.SignalExternalWorkflowExecutionFailedEventAttributes) *historypb.HistoryEvent {
 	return &historypb.HistoryEvent{
 		EventId:    eventID,
@@ -324,6 +331,33 @@ func createTestEventVersionMarker(eventID int64, workflowTaskCompletedID int64,
 	}
 }
 
+func createTestEventSideEffectMarker(eventID int64, workflowTaskCompletedID int64, sideEffectID int64, result int) *historypb.HistoryEvent {
+	sideEffectIDPayload, err := converter.GetDefaultDataConverter().ToPayloads(sideEffectID)
+	if err != nil {
+		panic(err)
+	}
+
+	resultPayload, err := converter.GetDefaultDataConverter().ToPayloads(result)
+	if err != nil {
+		panic(err)
+	}
+
+	return &historypb.HistoryEvent{
+		EventId:   eventID,
+		EventType: enumspb.EVENT_TYPE_MARKER_RECORDED,
+		Attributes: &historypb.HistoryEvent_MarkerRecordedEventAttributes{
+			MarkerRecordedEventAttributes: &historypb.MarkerRecordedEventAttributes{
+				MarkerName: sideEffectMarkerName,
+				Details: map[string]*commonpb.Payloads{
+					sideEffectMarkerIDName:   sideEffectIDPayload,
+					sideEffectMarkerDataName: resultPayload,
+				},
+				WorkflowTaskCompletedEventId: workflowTaskCompletedID,
+			},
+		},
+	}
+}
+
 func createTestUpsertWorkflowSearchAttributesForChangeVersion(eventID int64, workflowTaskCompletedID int64, changeID string, version Version) *historypb.HistoryEvent {
 	searchAttributes, _ := validateAndSerializeSearchAttributes(createSearchAttributesForChangeVersion(changeID, version, nil))
 

diff --git a/internal/internal_worker_test.go b/internal/internal_worker_test.go
@@ -1368,6 +1368,136 @@ func (s *internalWorkerTestSuite) TestReplayWorkflowHistory_LocalActivity_Activi
 	require.Error(s.T(), err)
 }
 
+func testReplayWorkflowSideEffect(ctx Context) error {
+	if err := Sleep(ctx, time.Second); err != nil {
+		return err
+	}
+
+	encodedRandom := SideEffect(ctx, func(ctx Context) interface{} {
+		return 100
+	})
+
+	var random int
+	if err := encodedRandom.Get(&random); err != nil {
+		return err
+	}
+	if random == 100 {
+		ao := ActivityOptions{
+			ScheduleToStartTimeout: time.Second,
+			StartToCloseTimeout:    time.Second,
+		}
+		ctx = WithActivityOptions(ctx, ao)
+
+		a1F := ExecuteActivity(ctx, "A1", "first")
+		a2F := ExecuteActivity(ctx, "A2", "second")
+		a3F := ExecuteActivity(ctx, "A3", "third")
+
+		var A1Result string
+		if err := a1F.Get(ctx, &A1Result); err != nil {
+			return err
+		}
+
+		err := a3F.Get(ctx, nil)
+		if err == nil {
+			getLogger().Info("activity A3 completed.")
+		}
+
+		err = a2F.Get(ctx, nil)
+		if err != nil {
+			getLogger().Info("activity A2 completed.")
+		}
+
+	}
+
+	getLogger().Info("workflow completed.")
+	return nil
+}
+
+func (s *internalWorkerTestSuite) TestReplayWorkflowHistory_SideEffect() {
+	taskQueue := "taskQueue1"
+	testEvents := []*historypb.HistoryEvent{
+		createTestEventWorkflowExecutionStarted(1, &historypb.WorkflowExecutionStartedEventAttributes{
+			WorkflowType: &commonpb.WorkflowType{Name: "testReplayWorkflowSideEffect"},
+			TaskQueue:    &taskqueuepb.TaskQueue{Name: taskQueue},
+			Input:        testEncodeFunctionArgs(converter.GetDefaultDataConverter()),
+		}),
+		createTestEventWorkflowTaskScheduled(2, &historypb.WorkflowTaskScheduledEventAttributes{}),
+		createTestEventWorkflowTaskStarted(3),
+		createTestEventWorkflowTaskCompleted(4, &historypb.WorkflowTaskCompletedEventAttributes{}),
+		createTestEventTimerStarted(5, 5),
+		createTestEventTimerFired(6, 5),
+		createTestEventWorkflowTaskScheduled(7, &historypb.WorkflowTaskScheduledEventAttributes{}),
+		createTestEventWorkflowTaskStarted(8),
+		createTestEventWorkflowTaskCompleted(9, &historypb.WorkflowTaskCompletedEventAttributes{}),
+		createTestEventSideEffectMarker(10, 9, 1, 100),
+
+		createTestEventActivityTaskScheduled(11, &historypb.ActivityTaskScheduledEventAttributes{
+			ActivityId:   "11",
+			ActivityType: &commonpb.ActivityType{Name: "A1"},
+			TaskQueue:    &taskqueuepb.TaskQueue{Name: taskQueue},
+		}),
+		createTestEventActivityTaskScheduled(12, &historypb.ActivityTaskScheduledEventAttributes{
+			ActivityId:   "12",
+			ActivityType: &commonpb.ActivityType{Name: "A2"},
+			TaskQueue:    &taskqueuepb.TaskQueue{Name: taskQueue},
+		}),
+		createTestEventActivityTaskScheduled(13, &historypb.ActivityTaskScheduledEventAttributes{
+			ActivityId:   "13",
+			ActivityType: &commonpb.ActivityType{Name: "A3"},
+			TaskQueue:    &taskqueuepb.TaskQueue{Name: taskQueue},
+		}),
+		createTestEventActivityTaskStarted(14, &historypb.ActivityTaskStartedEventAttributes{
+			ScheduledEventId: 11,
+		}),
+		createTestEventActivityTaskCompleted(15, &historypb.ActivityTaskCompletedEventAttributes{
+			ScheduledEventId: 11,
+			StartedEventId:   14,
+		}),
+		createTestEventWorkflowTaskScheduled(16, &historypb.WorkflowTaskScheduledEventAttributes{}),
+		createTestEventWorkflowTaskStarted(17),
+		createTestEventWorkflowTaskCompleted(18, &historypb.WorkflowTaskCompletedEventAttributes{
+			ScheduledEventId: 16,
+			StartedEventId:   17,
+		}),
+		createTestEventActivityTaskStarted(19, &historypb.ActivityTaskStartedEventAttributes{
+			ScheduledEventId: 13,
+		}),
+		createTestEventActivityTaskCompleted(20, &historypb.ActivityTaskCompletedEventAttributes{
+			ScheduledEventId: 13,
+			StartedEventId:   19,
+		}),
+		createTestEventWorkflowTaskScheduled(21, &historypb.WorkflowTaskScheduledEventAttributes{}),
+		createTestEventActivityTaskStarted(22, &historypb.ActivityTaskStartedEventAttributes{
+			ScheduledEventId: 12,
+		}),
+		createTestEventActivityTaskCompleted(23, &historypb.ActivityTaskCompletedEventAttributes{
+			ScheduledEventId: 12,
+			StartedEventId:   22,
+		}),
+		createTestEventWorkflowTaskTimedOut(24, &historypb.WorkflowTaskTimedOutEventAttributes{
+			ScheduledEventId: 21,
+			StartedEventId:   0,
+			TimeoutType:      enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
+		}),
+		createTestEventWorkflowTaskScheduled(25, &historypb.WorkflowTaskScheduledEventAttributes{}),
+		createTestEventWorkflowTaskStarted(26),
+		createTestEventWorkflowTaskCompleted(27, &historypb.WorkflowTaskCompletedEventAttributes{
+			ScheduledEventId: 25,
+			StartedEventId:   26,
+		}),
+		createTestEventWorkflowExecutionCompleted(28, &historypb.WorkflowExecutionCompletedEventAttributes{
+			WorkflowTaskCompletedEventId: 27,
+		}),
+	}
+
+	history := &historypb.History{Events: testEvents}
+	logger := getLogger()
+	replayer := NewWorkflowReplayer()
+	replayer.RegisterWorkflow(testReplayWorkflowSideEffect)
+	err := replayer.ReplayWorkflowHistory(logger, history)
+	require.NoError(s.T(), err)
+}
+
 func (s *internalWorkerTestSuite) TestReplayWorkflowHistoryFromFileParent() {
 	logger := getLogger()
 	replayer := NewWorkflowReplayer()