Truncate history for query based on nextEventID (#1096)

Author: yycptt

.gen/go/shared/shared.go

@@ -42,6 +42,7 @@ const (
 	DecisionPollNoTaskCounter          = CadenceMetricsPrefix + "decision-poll-no-task"
 	DecisionPollSucceedCounter         = CadenceMetricsPrefix + "decision-poll-succeed"
 	DecisionPollLatency                = CadenceMetricsPrefix + "decision-poll-latency" // measure succeed poll request latency
+	DecisionPollInvalidCounter         = CadenceMetricsPrefix + "decision-poll-invalid"
 	DecisionScheduledToStartLatency    = CadenceMetricsPrefix + "decision-scheduled-to-start-latency"
 	DecisionExecutionFailedCounter     = CadenceMetricsPrefix + "decision-execution-failed"
 	DecisionExecutionLatency           = CadenceMetricsPrefix + "decision-execution-latency"

idls

@@ -169,7 +169,11 @@ func newHistory(task *workflowTask, eventsHandler *workflowExecutionEventHandler
 		eventsHandler: eventsHandler,
 		loadedEvents:  task.task.History.Events,
 		currentIndex:  0,
-		lastEventID:   task.task.GetStartedEventId(),
+		// don't set lastEventID to task.GetNextEventId()
+		// as for sticky query, the history in workflow task will be empty
+		// and query will be run based on existing workflow state.
+		// so the sanity check in verifyAllEventsProcessed will fail
+		lastEventID: task.task.GetStartedEventId(),
 	}
 	if len(result.loadedEvents) > 0 {
 		result.nextEventID = result.loadedEvents[0].GetEventId()

internal/common/metrics/constants.go

@@ -487,6 +487,9 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_QueryWorkflow_NonSticky() {
 		Logger:   t.logger,
 	}
 
+	// TODO: the following comment is not true, previousStartEventID is either 0 or points to a valid decisionTaskStartedID
+	// we need to fix the test
+	//
 	// query after first decision task (notice the previousStartEventID is always the last eventID for query task)
 	task := createQueryTask(testEvents[0:3], 3, "HelloWorld_Workflow", queryType)
 	taskHandler := newWorkflowTaskHandler(testDomain, params, nil, t.registry)

internal/internal_task_handlers.go

@@ -105,13 +105,14 @@ type (
 	}
 
 	historyIteratorImpl struct {
-		iteratorFunc  func(nextPageToken []byte) (*s.History, []byte, error)
-		execution     *s.WorkflowExecution
-		nextPageToken []byte
-		domain        string
-		service       workflowserviceclient.Interface
-		metricsScope  tally.Scope
-		maxEventID    int64
+		iteratorFunc   func(nextPageToken []byte) (*s.History, []byte, error)
+		execution      *s.WorkflowExecution
+		nextPageToken  []byte
+		domain         string
+		service        workflowserviceclient.Interface
+		metricsScope   tally.Scope
+		startedEventID int64
+		maxEventID     int64
 	}
 
 	localActivityTaskPoller struct {
@@ -798,13 +799,25 @@ func (wtp *workflowTaskPoller) poll(ctx context.Context) (interface{}, error) {
 }
 
 func (wtp *workflowTaskPoller) toWorkflowTask(response *s.PollForDecisionTaskResponse) *workflowTask {
+	startEventID := response.GetStartedEventId()
+	nextEventID := response.GetNextEventId()
+	if nextEventID != 0 &&
+		startEventID != 0 &&
+		nextEventID-1 != startEventID {
+		wtp.logger.Warn("Invalid PollForDecisionTaskResponse, nextEventID doesn't match startedEventID",
+			zap.Int64("StartedEventID", startEventID),
+			zap.Int64("NextEventID", nextEventID),
+		)
+		wtp.metricsScope.Counter(metrics.DecisionPollInvalidCounter).Inc(1)
+	}
 	historyIterator := &historyIteratorImpl{
-		nextPageToken: response.NextPageToken,
-		execution:     response.WorkflowExecution,
-		domain:        wtp.domain,
-		service:       wtp.service,
-		metricsScope:  wtp.metricsScope,
-		maxEventID:    response.GetStartedEventId(),
+		nextPageToken:  response.NextPageToken,
+		execution:      response.WorkflowExecution,
+		domain:         wtp.domain,
+		service:        wtp.service,
+		metricsScope:   wtp.metricsScope,
+		startedEventID: startEventID,
+		maxEventID:     nextEventID - 1,
 	}
 	task := &workflowTask{
 		task:            response,
@@ -820,6 +833,7 @@ func (h *historyIteratorImpl) GetNextPage() (*s.History, error) {
 			h.service,
 			h.domain,
 			h.execution,
+			h.startedEventID,
 			h.maxEventID,
 			h.metricsScope)
 	}
@@ -846,6 +860,7 @@ func newGetHistoryPageFunc(
 	domain string,
 	execution *s.WorkflowExecution,
 	atDecisionTaskCompletedEventID int64,
+	maxEventID int64,
 	metricsScope tally.Scope,
 ) func(nextPageToken []byte) (*s.History, []byte, error) {
 	return func(nextPageToken []byte) (*s.History, []byte, error) {
@@ -885,9 +900,12 @@ func newGetHistoryPageFunc(
 			h = resp.History
 		}
 
-		size := len(h.Events)
-		if size > 0 && atDecisionTaskCompletedEventID > 0 &&
-			h.Events[size-1].GetEventId() > atDecisionTaskCompletedEventID {
+		// TODO: is this check valid/useful? atDecisionTaskCompletedEventID is startedEventID in pollForDecisionTaskResponse and
+		// - For decision tasks, since there's only one inflight decision task, there won't be any event after startEventID.
+		//   Those events will be buffered. If there're too many buffer events, the current decision will be failed and events passed
+		//   startEventID may be returned. In that case, the last event after truncation is still decision task started event not completed.
+		// - For query tasks startEventID is not assigned so this check is never executed.
+		if shouldTruncateHistory(h, atDecisionTaskCompletedEventID) {
 			first := h.Events[0].GetEventId() // eventIds start from 1
 			h.Events = h.Events[:atDecisionTaskCompletedEventID-first+1]
 			if h.Events[len(h.Events)-1].GetEventType() != s.EventTypeDecisionTaskCompleted {
@@ -896,10 +914,29 @@ func newGetHistoryPageFunc(
 			}
 			return h, nil, nil
 		}
+
+		// TODO: Apply the check to decision tasks (remove the last condition)
+		// after validating maxEventID always equal to atDecisionTaskCompletedEventID (startedEventID).
+		// For now only apply to query task to be safe.
+		if shouldTruncateHistory(h, maxEventID) && isQueryTask(atDecisionTaskCompletedEventID) {
+			first := h.Events[0].GetEventId()
+			h.Events = h.Events[:maxEventID-first+1]
+			return h, nil, nil
+		}
+
 		return h, resp.NextPageToken, nil
 	}
 }
 
+func shouldTruncateHistory(h *s.History, maxEventID int64) bool {
+	size := len(h.Events)
+	return size > 0 && maxEventID > 0 && h.Events[size-1].GetEventId() > maxEventID
+}
+
+func isQueryTask(atDecisionTaskCompletedEventID int64) bool {
+	return atDecisionTaskCompletedEventID == 0
+}
+
 func newActivityTaskPoller(taskHandler ActivityTaskHandler, service workflowserviceclient.Interface,
 	domain string, params workerExecutionParameters) *activityTaskPoller {
 

internal/internal_task_handlers_test.go

@@ -304,6 +304,7 @@ func (s *WorkersTestSuite) TestLongRunningDecisionTask() {
 		StartedEventId:         common.Int64Ptr(3),
 		History:                &m.History{Events: testEvents[0:3]},
 		NextPageToken:          nil,
+		NextEventId:            common.Int64Ptr(4),
 	}
 	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(task, nil).Times(1)
 	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(&m.PollForDecisionTaskResponse{}, &m.InternalServiceError{}).AnyTimes()
@@ -359,6 +360,174 @@ func (s *WorkersTestSuite) TestLongRunningDecisionTask() {
 	s.Equal(2, localActivityCalledCount)
 }
 
+func (s *WorkersTestSuite) TestQueryTask_WorkflowCacheEvicted() {
+	domain := "testDomain"
+	taskList := "query-task-cache-evicted-tl"
+	workflowType := "query-task-cache-evicted-workflow"
+	workflowID := "query-task-cache-evicted-workflow-id"
+	runID := "query-task-cache-evicted-workflow-run-id"
+	activityType := "query-task-cache-evicted-activity"
+	queryType := "state"
+	doneCh := make(chan struct{})
+
+	activityFn := func(ctx context.Context) error {
+		return nil
+	}
+
+	queryWorkflowFn := func(ctx Context) error {
+		queryResult := "started"
+		// setup query handler for query type "state"
+		if err := SetQueryHandler(ctx, queryType, func(input []byte) (string, error) {
+			return queryResult, nil
+		}); err != nil {
+			return err
+		}
+
+		queryResult = "waiting on timer"
+		NewTimer(ctx, time.Minute*2).Get(ctx, nil)
+
+		queryResult = "waiting on activity"
+		ctx = WithActivityOptions(ctx, ActivityOptions{
+			ScheduleToStartTimeout: 10 * time.Second,
+			StartToCloseTimeout:    10 * time.Second,
+		})
+		if err := ExecuteActivity(ctx, activityFn).Get(ctx, nil); err != nil {
+			return err
+		}
+		queryResult = "done"
+		return nil
+	}
+
+	testEvents := []*m.HistoryEvent{
+		{
+			EventId:   common.Int64Ptr(1),
+			EventType: common.EventTypePtr(m.EventTypeWorkflowExecutionStarted),
+			WorkflowExecutionStartedEventAttributes: &m.WorkflowExecutionStartedEventAttributes{
+				TaskList:                            &m.TaskList{Name: &taskList},
+				ExecutionStartToCloseTimeoutSeconds: common.Int32Ptr(180),
+				TaskStartToCloseTimeoutSeconds:      common.Int32Ptr(2),
+				WorkflowType:                        &m.WorkflowType{Name: common.StringPtr(workflowType)},
+			},
+		},
+		createTestEventDecisionTaskScheduled(2, &m.DecisionTaskScheduledEventAttributes{TaskList: &m.TaskList{Name: &taskList}}),
+		createTestEventDecisionTaskStarted(3),
+		createTestEventDecisionTaskCompleted(4, &m.DecisionTaskCompletedEventAttributes{ScheduledEventId: common.Int64Ptr(2)}),
+		{
+			EventId:   common.Int64Ptr(5),
+			EventType: common.EventTypePtr(m.EventTypeTimerStarted),
+			TimerStartedEventAttributes: &m.TimerStartedEventAttributes{
+				TimerId:                      common.StringPtr("0"),
+				StartToFireTimeoutSeconds:    common.Int64Ptr(120),
+				DecisionTaskCompletedEventId: common.Int64Ptr(4),
+			},
+		},
+		{
+			EventId:   common.Int64Ptr(6),
+			EventType: common.EventTypePtr(m.EventTypeTimerFired),
+			TimerFiredEventAttributes: &m.TimerFiredEventAttributes{
+				TimerId:        common.StringPtr("0"),
+				StartedEventId: common.Int64Ptr(5),
+			},
+		},
+		createTestEventDecisionTaskScheduled(7, &m.DecisionTaskScheduledEventAttributes{TaskList: &m.TaskList{Name: &taskList}}),
+		createTestEventDecisionTaskStarted(8),
+		createTestEventDecisionTaskCompleted(9, &m.DecisionTaskCompletedEventAttributes{ScheduledEventId: common.Int64Ptr(2)}),
+		createTestEventActivityTaskScheduled(10, &m.ActivityTaskScheduledEventAttributes{
+			ActivityId: common.StringPtr("1"),
+			ActivityType: &m.ActivityType{
+				Name: common.StringPtr(activityType),
+			},
+			Domain: common.StringPtr(domain),
+			TaskList: &m.TaskList{
+				Name: common.StringPtr(taskList),
+			},
+			ScheduleToStartTimeoutSeconds: common.Int32Ptr(10),
+			StartToCloseTimeoutSeconds:    common.Int32Ptr(10),
+			DecisionTaskCompletedEventId:  common.Int64Ptr(9),
+		}),
+	}
+
+	s.service.EXPECT().DescribeDomain(gomock.Any(), gomock.Any(), callOptions...).Return(nil, nil).AnyTimes()
+	task := &m.PollForDecisionTaskResponse{
+		TaskToken: []byte("test-token"),
+		WorkflowExecution: &m.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+		WorkflowType: &m.WorkflowType{
+			Name: common.StringPtr(workflowType),
+		},
+		PreviousStartedEventId: common.Int64Ptr(0),
+		StartedEventId:         common.Int64Ptr(3),
+		History:                &m.History{Events: testEvents[0:3]},
+		NextPageToken:          nil,
+		NextEventId:            common.Int64Ptr(4),
+	}
+	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(task, nil).Times(1)
+	s.service.EXPECT().RespondDecisionTaskCompleted(gomock.Any(), gomock.Any(), callOptions...).DoAndReturn(func(ctx context.Context, request *m.RespondDecisionTaskCompletedRequest, opts ...yarpc.CallOption,
+	) (success *m.RespondDecisionTaskCompletedResponse, err error) {
+		s.Equal(1, len(request.Decisions))
+		s.Equal(m.DecisionTypeStartTimer, request.Decisions[0].GetDecisionType())
+		return &m.RespondDecisionTaskCompletedResponse{}, nil
+	}).Times(1)
+	queryTask := &m.PollForDecisionTaskResponse{
+		TaskToken: []byte("test-token"),
+		WorkflowExecution: &m.WorkflowExecution{
+			WorkflowId: common.StringPtr(workflowID),
+			RunId:      common.StringPtr(runID),
+		},
+		WorkflowType: &m.WorkflowType{
+			Name: common.StringPtr(workflowType),
+		},
+		PreviousStartedEventId: common.Int64Ptr(3),
+		History:                &m.History{}, // sticky query, so there's no history
+		NextPageToken:          nil,
+		NextEventId:            common.Int64Ptr(5),
+		Query: &m.WorkflowQuery{
+			QueryType: common.StringPtr(queryType),
+		},
+	}
+	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).DoAndReturn(func(ctx context.Context, request *m.PollForDecisionTaskRequest, opts ...yarpc.CallOption,
+	) (success *m.PollForDecisionTaskResponse, err error) {
+		getWorkflowCache().Delete(runID) // force remove the workflow state
+		return queryTask, nil
+	}).Times(1)
+	s.service.EXPECT().ResetStickyTaskList(gomock.Any(), gomock.Any(), callOptions...).Return(&m.ResetStickyTaskListResponse{}, nil).AnyTimes()
+	s.service.EXPECT().GetWorkflowExecutionHistory(gomock.Any(), gomock.Any(), callOptions...).Return(&m.GetWorkflowExecutionHistoryResponse{
+		History: &m.History{Events: testEvents}, // workflow has made progress, return all available events
+	}, nil).Times(1)
+	dc := getDefaultDataConverter()
+	expectedResult, err := dc.ToData("waiting on timer")
+	s.NoError(err)
+	s.service.EXPECT().RespondQueryTaskCompleted(gomock.Any(), gomock.Any(), callOptions...).DoAndReturn(func(ctx context.Context, request *m.RespondQueryTaskCompletedRequest, opts ...yarpc.CallOption) error {
+		s.Equal(m.QueryTaskCompletedTypeCompleted, request.GetCompletedType())
+		s.Equal(expectedResult, request.GetQueryResult())
+		close(doneCh)
+		return nil
+	}).Times(1)
+	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(&m.PollForDecisionTaskResponse{}, &m.InternalServiceError{}).AnyTimes()
+
+	options := WorkerOptions{
+		Logger:                zap.NewNop(),
+		DisableActivityWorker: true,
+		Identity:              "test-worker-identity",
+		DataConverter:         dc,
+	}
+	worker := newAggregatedWorker(s.service, domain, taskList, options)
+	worker.RegisterWorkflowWithOptions(
+		queryWorkflowFn,
+		RegisterWorkflowOptions{Name: workflowType},
+	)
+	worker.RegisterActivityWithOptions(
+		activityFn,
+		RegisterActivityOptions{Name: activityType},
+	)
+
+	worker.Start()
+	<-doneCh
+	worker.Stop()
+}
+
 func (s *WorkersTestSuite) TestMultipleLocalActivities() {
 	localActivityCalledCount := 0
 	localActivitySleep := func(duration time.Duration) error {
@@ -441,6 +610,7 @@ func (s *WorkersTestSuite) TestMultipleLocalActivities() {
 		StartedEventId:         common.Int64Ptr(3),
 		History:                &m.History{Events: testEvents[0:3]},
 		NextPageToken:          nil,
+		NextEventId:            common.Int64Ptr(4),
 	}
 	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(task, nil).Times(1)
 	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(&m.PollForDecisionTaskResponse{}, &m.InternalServiceError{}).AnyTimes()
@@ -554,6 +724,7 @@ func (s *WorkersTestSuite) TestLocallyDispatchedActivity() {
 		StartedEventId:         common.Int64Ptr(3),
 		History:                &m.History{Events: testEvents[0:3]},
 		NextPageToken:          nil,
+		NextEventId:            common.Int64Ptr(4),
 	}
 	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(task, nil).Times(1)
 	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(&m.PollForDecisionTaskResponse{}, &m.InternalServiceError{}).AnyTimes()
@@ -666,6 +837,7 @@ func (s *WorkersTestSuite) TestMultipleLocallyDispatchedActivity() {
 		StartedEventId:         common.Int64Ptr(3),
 		History:                &m.History{Events: testEvents[0:3]},
 		NextPageToken:          nil,
+		NextEventId:            common.Int64Ptr(4),
 	}
 	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(task, nil).Times(1)
 	s.service.EXPECT().PollForDecisionTask(gomock.Any(), gomock.Any(), callOptions...).Return(&m.PollForDecisionTaskResponse{}, &m.InternalServiceError{}).AnyTimes()

internal/internal_task_pollers.go

@@ -276,12 +276,12 @@ func (r *WorkflowReplayer) replayWorkflowHistory(
 
 	metricScope := tally.NoopScope
 	iterator := &historyIteratorImpl{
-		nextPageToken: task.NextPageToken,
-		execution:     task.WorkflowExecution,
-		domain:        domain,
-		service:       service,
-		metricsScope:  metricScope,
-		maxEventID:    task.GetStartedEventId(),
+		nextPageToken:  task.NextPageToken,
+		execution:      task.WorkflowExecution,
+		domain:         domain,
+		service:        service,
+		metricsScope:   metricScope,
+		startedEventID: task.GetStartedEventId(),
 	}
 	taskHandler := newWorkflowTaskHandler(domain, workerParams, nil, r.registry)
 	resp, err := taskHandler.ProcessWorkflowTask(&workflowTask{task: task, historyIterator: iterator}, nil)