Change visibility closed_time to use last event timestamp (cadence-wo…

…rkflow#1501)

* Change visibility closed_time to use last event timestamp

* Move writeCompletionEventToCache to replicate APIs

* Remove GetLastUpdatedTimestamp API

schema/cassandra/cadence/schema.cql

@@ -310,7 +310,7 @@ CREATE TABLE executions (
     'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
   };
 
--- events table is deprecated in favor of v2 talbes: history_node/history_tree
+-- events table is deprecated in favor of v2 tables: history_node/history_tree
 CREATE TABLE events (
   domain_id           uuid,
   workflow_id         text,

service/history/MockMutableState.go

@@ -1348,20 +1348,6 @@ func (_m *mockMutableState) GetLastFirstEventID() int64 {
 	return r0
 }
 
-// GetLastUpdatedTimestamp provides a mock function with given fields:
-func (_m *mockMutableState) GetLastUpdatedTimestamp() int64 {
-	ret := _m.Called()
-
-	var r0 int64
-	if rf, ok := ret.Get(0).(func() int64); ok {
-		r0 = rf()
-	} else {
-		r0 = ret.Get(0).(int64)
-	}
-
-	return r0
-}
-
 // GetLastWriteVersion provides a mock function with given fields:
 func (_m *mockMutableState) GetLastWriteVersion() int64 {
 	ret := _m.Called()

service/history/historyEngine.go

@@ -819,7 +819,11 @@ func (e *historyEngineImpl) DescribeWorkflowExecution(ctx context.Context,
 		// for closed workflow
 		closeStatus := getWorkflowExecutionCloseStatus(executionInfo.CloseStatus)
 		result.WorkflowExecutionInfo.CloseStatus = &closeStatus
-		result.WorkflowExecutionInfo.CloseTime = common.Int64Ptr(msBuilder.GetLastUpdatedTimestamp())
+		completionEvent, ok := msBuilder.GetCompletionEvent()
+		if !ok {
+			return nil, &workflow.InternalServiceError{Message: "Unable to get workflow completion event."}
+		}
+		result.WorkflowExecutionInfo.CloseTime = common.Int64Ptr(completionEvent.GetTimestamp())
 	}
 
 	if len(msBuilder.GetPendingActivityInfos()) > 0 {
@@ -1283,7 +1287,7 @@ Update_History_Loop:
 					}
 
 					startAttributes := startEvent.WorkflowExecutionStartedEventAttributes
-					continueAsnewAttributes := &workflow.ContinueAsNewWorkflowExecutionDecisionAttributes{
+					continueAsNewAttributes := &workflow.ContinueAsNewWorkflowExecutionDecisionAttributes{
 						WorkflowType:                        startAttributes.WorkflowType,
 						TaskList:                            startAttributes.TaskList,
 						RetryPolicy:                         startAttributes.RetryPolicy,
@@ -1301,7 +1305,7 @@ Update_History_Loop:
 						return nil, err
 					}
 					if _, continueAsNewBuilder, err = msBuilder.AddContinueAsNewEvent(completedID, domainEntry,
-						startAttributes.GetParentWorkflowDomain(), continueAsnewAttributes, eventStoreVersion,
+						startAttributes.GetParentWorkflowDomain(), continueAsNewAttributes, eventStoreVersion,
 						createTaskID); err != nil {
 						return nil, err
 					}
@@ -1361,7 +1365,7 @@ Update_History_Loop:
 					}
 
 					startAttributes := startEvent.WorkflowExecutionStartedEventAttributes
-					continueAsnewAttributes := &workflow.ContinueAsNewWorkflowExecutionDecisionAttributes{
+					continueAsNewAttributes := &workflow.ContinueAsNewWorkflowExecutionDecisionAttributes{
 						WorkflowType:                        startAttributes.WorkflowType,
 						TaskList:                            startAttributes.TaskList,
 						RetryPolicy:                         startAttributes.RetryPolicy,
@@ -1382,7 +1386,7 @@ Update_History_Loop:
 					}
 					if _, continueAsNewBuilder, err = msBuilder.AddContinueAsNewEvent(completedID, domainEntry,
 						startAttributes.GetParentWorkflowDomain(),
-						continueAsnewAttributes, eventStoreVersion, createTaskID); err != nil {
+						continueAsNewAttributes, eventStoreVersion, createTaskID); err != nil {
 						return nil, err
 					}
 				}

service/history/historyReplicator.go

@@ -686,7 +686,7 @@ func (r *historyReplicator) ApplyReplicationTask(ctx context.Context, context wo
 		if err != nil {
 			return err
 		}
-		// contineueAsNew
+		// continueAsNew
 		err = context.appendFirstBatchHistoryForContinueAsNew(newRunStateBuilder, transactionID)
 		if err != nil {
 			return err

service/history/mutableState.go

@@ -123,7 +123,6 @@ type (
 		GetHistorySize() int64
 		GetInFlightDecisionTask() (*decisionInfo, bool)
 		GetLastFirstEventID() int64
-		GetLastUpdatedTimestamp() int64
 		GetLastWriteVersion() int64
 		GetNextEventID() int64
 		GetPreviousStartedEventID() int64

service/history/mutableStateBuilder.go

@@ -772,17 +772,6 @@ func (e *mutableStateBuilder) GetWorkflowType() *workflow.WorkflowType {
 	return wType
 }
 
-func (e *mutableStateBuilder) GetLastUpdatedTimestamp() int64 {
-	lastUpdated := e.executionInfo.LastUpdatedTimestamp.UnixNano()
-	if e.executionInfo.StartTimestamp.UnixNano() >= lastUpdated {
-		// This could happen due to clock skews
-		// ensure that the lastUpdatedTimestamp is always greater than the StartTimestamp
-		lastUpdated = e.executionInfo.StartTimestamp.UnixNano() + 1
-	}
-
-	return lastUpdated
-}
-
 func (e *mutableStateBuilder) GetActivityScheduledEvent(scheduleEventID int64) (*workflow.HistoryEvent, bool) {
 	ai, ok := e.pendingActivityInfoIDs[scheduleEventID]
 	if !ok {
@@ -933,16 +922,11 @@ func (e *mutableStateBuilder) DeletePendingSignal(initiatedEventID int64) {
 	e.deleteSignalInfo = common.Int64Ptr(initiatedEventID)
 }
 
-func (e *mutableStateBuilder) writeCompletionEventToCache(completionEvent *workflow.HistoryEvent) error {
-	// First check to see if this is a Child Workflow
-	if e.HasParentExecution() {
-		// Store the completion result within events cache so we can communicate the result to parent execution
-		// during the processing of DeleteTransferTask without loading this event from database
-		e.eventsCache.putEvent(e.executionInfo.DomainID, e.executionInfo.WorkflowID, e.executionInfo.RunID,
-			completionEvent.GetEventId(), completionEvent)
-	}
-
-	return nil
+func (e *mutableStateBuilder) writeCompletionEventToCache(completionEvent *workflow.HistoryEvent) {
+	// Store the completion result within events cache so we can communicate the result to parent execution
+	// during the processing of DeleteTransferTask without loading this event from database
+	e.eventsCache.putEvent(e.executionInfo.DomainID, e.executionInfo.WorkflowID, e.executionInfo.RunID,
+		completionEvent.GetEventId(), completionEvent)
 }
 
 func (e *mutableStateBuilder) hasPendingTasks() bool {
@@ -1917,17 +1901,16 @@ func (e *mutableStateBuilder) AddCompletedWorkflowEvent(decisionCompletedEventID
 	}
 
 	event := e.hBuilder.AddCompletedWorkflowEvent(decisionCompletedEventID, attributes)
-	// Write the event to cache only on active cluster
-	e.writeCompletionEventToCache(event)
-
 	e.ReplicateWorkflowExecutionCompletedEvent(decisionCompletedEventID, event)
+
 	return event
 }
 
 func (e *mutableStateBuilder) ReplicateWorkflowExecutionCompletedEvent(firstEventID int64, event *workflow.HistoryEvent) {
 	e.executionInfo.State = persistence.WorkflowStateCompleted
 	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusCompleted
 	e.executionInfo.CompletionEventBatchID = firstEventID // Used when completion event needs to be loaded from database
+	e.writeCompletionEventToCache(event)
 }
 
 func (e *mutableStateBuilder) AddFailWorkflowEvent(decisionCompletedEventID int64,
@@ -1939,9 +1922,6 @@ func (e *mutableStateBuilder) AddFailWorkflowEvent(decisionCompletedEventID int6
 	}
 
 	event := e.hBuilder.AddFailWorkflowEvent(decisionCompletedEventID, attributes)
-	// Write the event to cache only on active cluster
-	e.writeCompletionEventToCache(event)
-
 	e.ReplicateWorkflowExecutionFailedEvent(decisionCompletedEventID, event)
 
 	return event
@@ -1951,6 +1931,7 @@ func (e *mutableStateBuilder) ReplicateWorkflowExecutionFailedEvent(firstEventID
 	e.executionInfo.State = persistence.WorkflowStateCompleted
 	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusFailed
 	e.executionInfo.CompletionEventBatchID = firstEventID // Used when completion event needs to be loaded from database
+	e.writeCompletionEventToCache(event)
 }
 
 func (e *mutableStateBuilder) AddTimeoutWorkflowEvent() *workflow.HistoryEvent {
@@ -1961,9 +1942,6 @@ func (e *mutableStateBuilder) AddTimeoutWorkflowEvent() *workflow.HistoryEvent {
 	}
 
 	event := e.hBuilder.AddTimeoutWorkflowEvent()
-	// Write the event to cache only on active cluster
-	e.writeCompletionEventToCache(event)
-
 	e.ReplicateWorkflowExecutionTimedoutEvent(event.GetEventId(), event)
 
 	return event
@@ -1973,6 +1951,7 @@ func (e *mutableStateBuilder) ReplicateWorkflowExecutionTimedoutEvent(firstEvent
 	e.executionInfo.State = persistence.WorkflowStateCompleted
 	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusTimedOut
 	e.executionInfo.CompletionEventBatchID = firstEventID // Used when completion event needs to be loaded from database
+	e.writeCompletionEventToCache(event)
 }
 
 func (e *mutableStateBuilder) AddWorkflowExecutionCancelRequestedEvent(cause string,
@@ -2005,8 +1984,6 @@ func (e *mutableStateBuilder) AddWorkflowExecutionCanceledEvent(decisionTaskComp
 	}
 
 	event := e.hBuilder.AddWorkflowExecutionCanceledEvent(decisionTaskCompletedEventID, attributes)
-	// Write the event to cache only on active cluster
-	e.writeCompletionEventToCache(event)
 	e.ReplicateWorkflowExecutionCanceledEvent(decisionTaskCompletedEventID, event)
 
 	return event
@@ -2016,6 +1993,7 @@ func (e *mutableStateBuilder) ReplicateWorkflowExecutionCanceledEvent(firstEvent
 	e.executionInfo.State = persistence.WorkflowStateCompleted
 	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusCanceled
 	e.executionInfo.CompletionEventBatchID = firstEventID // Used when completion event needs to be loaded from database
+	e.writeCompletionEventToCache(event)
 }
 
 func (e *mutableStateBuilder) AddRequestCancelExternalWorkflowExecutionInitiatedEvent(
@@ -2272,9 +2250,6 @@ func (e *mutableStateBuilder) AddWorkflowExecutionTerminatedEvent(
 	}
 
 	event := e.hBuilder.AddWorkflowExecutionTerminatedEvent(request)
-	// Write the event to cache only on active cluster
-	e.writeCompletionEventToCache(event)
-
 	e.ReplicateWorkflowExecutionTerminatedEvent(event.GetEventId(), event)
 
 	return event
@@ -2284,6 +2259,7 @@ func (e *mutableStateBuilder) ReplicateWorkflowExecutionTerminatedEvent(firstEve
 	e.executionInfo.State = persistence.WorkflowStateCompleted
 	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusTerminated
 	e.executionInfo.CompletionEventBatchID = firstEventID // Used when completion event needs to be loaded from database
+	e.writeCompletionEventToCache(event)
 }
 
 func (e *mutableStateBuilder) AddWorkflowExecutionSignaled(
@@ -2364,6 +2340,9 @@ func (e *mutableStateBuilder) AddContinueAsNewEvent(decisionCompletedEventID int
 func (e *mutableStateBuilder) ReplicateWorkflowExecutionContinuedAsNewEvent(sourceClusterName string, domainID string,
 	continueAsNewEvent *workflow.HistoryEvent, startedEvent *workflow.HistoryEvent, di *decisionInfo,
 	newStateBuilder mutableState, eventStoreVersion int32, createTaskID int64) error {
+
+	e.writeCompletionEventToCache(continueAsNewEvent)
+
 	continueAsNewAttributes := continueAsNewEvent.WorkflowExecutionContinuedAsNewEventAttributes
 	startedAttributes := startedEvent.WorkflowExecutionStartedEventAttributes
 	newRunID := continueAsNewAttributes.GetNewExecutionRunId()

service/history/stateBuilder.go

@@ -109,7 +109,7 @@ func (b *stateBuilderImpl) applyEvents(domainID, requestID string, execution sha
 		firstEvent = history[0]
 	}
 
-	// need to clear the stickness since workflow turned to passive
+	// need to clear the stickiness since workflow turned to passive
 	b.msBuilder.ClearStickyness()
 	for _, event := range history {
 		lastEvent = event
@@ -146,7 +146,7 @@ func (b *stateBuilderImpl) applyEvents(domainID, requestID string, execution sha
 
 			b.transferTasks = append(b.transferTasks, b.scheduleDecisionTransferTask(domainID, b.getTaskList(b.msBuilder),
 				di.ScheduleID))
-			// since we do not use stickyness on the standby side, there shall be no decision schedule to start timeout
+			// since we do not use stickiness on the standby side, there shall be no decision schedule to start timeout
 
 			lastDecision = di
 
@@ -173,7 +173,7 @@ func (b *stateBuilderImpl) applyEvents(domainID, requestID string, execution sha
 			if di := b.msBuilder.ReplicateTransientDecisionTaskScheduled(); di != nil {
 				b.transferTasks = append(b.transferTasks, b.scheduleDecisionTransferTask(domainID, b.getTaskList(b.msBuilder),
 					di.ScheduleID))
-				// since we do not use stickyness on the standby side, there shall be no decision schedule to start timeout
+				// since we do not use stickiness on the standby side, there shall be no decision schedule to start timeout
 				lastDecision = di
 			}
 
@@ -184,7 +184,7 @@ func (b *stateBuilderImpl) applyEvents(domainID, requestID string, execution sha
 			if di := b.msBuilder.ReplicateTransientDecisionTaskScheduled(); di != nil {
 				b.transferTasks = append(b.transferTasks, b.scheduleDecisionTransferTask(domainID, b.getTaskList(b.msBuilder),
 					di.ScheduleID))
-				// since we do not use stickyness on the standby side, there shall be no decision schedule to start timeout
+				// since we do not use stickiness on the standby side, there shall be no decision schedule to start timeout
 				lastDecision = di
 			}
 

service/history/stateBuilder_test.go

@@ -59,6 +59,7 @@ type (
 		mockShard           *shardContextImpl
 		mockMutableState    *mockMutableState
 		mockClientBean      *client.MockClientBean
+		mockEventsCache     *MockEventsCache
 
 		stateBuilder *stateBuilderImpl
 	}
@@ -94,6 +95,7 @@ func (s *stateBuilderSuite) SetupTest() {
 	metricsClient := metrics.NewClient(tally.NoopScope, metrics.History)
 	s.mockClientBean = &client.MockClientBean{}
 	s.mockService = service.NewTestService(s.mockClusterMetadata, s.mockMessagingClient, metricsClient, s.mockClientBean, s.logger)
+	s.mockEventsCache = &MockEventsCache{}
 
 	s.mockShard = &shardContextImpl{
 		service:                   s.mockService,
@@ -107,6 +109,7 @@ func (s *stateBuilderSuite) SetupTest() {
 		config:                    NewDynamicConfigForTest(),
 		logger:                    s.logger,
 		domainCache:               cache.NewDomainCache(s.mockMetadataMgr, s.mockClusterMetadata, metricsClient, s.logger),
+		eventsCache:               s.mockEventsCache,
 		metricsClient:             metrics.NewClient(tally.NoopScope, metrics.History),
 	}
 	s.mockMutableState = &mockMutableState{}
@@ -124,6 +127,7 @@ func (s *stateBuilderSuite) TearDownTest() {
 	s.mockProducer.AssertExpectations(s.T())
 	s.mockMetadataMgr.AssertExpectations(s.T())
 	s.mockClientBean.AssertExpectations(s.T())
+	s.mockEventsCache.AssertExpectations(s.T())
 }
 
 func (s *stateBuilderSuite) mockUpdateVersion(events ...*shared.HistoryEvent) {
@@ -832,6 +836,8 @@ func (s *stateBuilderSuite) TestApplyEvents_EventTypeWorkflowExecutionContinuedA
 
 	newRunHistory := &shared.History{Events: []*shared.HistoryEvent{newRunStartedEvent, newRunSignalEvent, newRunDecisionEvent}}
 	s.mockMutableState.On("ClearStickyness").Once()
+	s.mockEventsCache.On("putEvent", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return().Once()
+
 	_, _, newRunStateBuilder, err := stateBuilder.applyEvents(domainID, requestID, execution, s.toHistory(continueAsNewEvent), newRunHistory.Events,
 		0, persistence.EventStoreVersionV2, createTaskID, newRunCreateTaskID)
 	s.Nil(err)

service/history/timerQueueProcessor2_test.go

@@ -166,6 +166,7 @@ func (s *timerQueueProcessor2Suite) TearDownTest() {
 	s.mockVisibilityMgr.AssertExpectations(s.T())
 	s.mockProducer.AssertExpectations(s.T())
 	s.mockClientBean.AssertExpectations(s.T())
+	s.mockEventsCache.AssertExpectations(s.T())
 }
 
 func (s *timerQueueProcessor2Suite) TestTimerUpdateTimesOut() {
@@ -286,6 +287,7 @@ func (s *timerQueueProcessor2Suite) TestWorkflowTimeout() {
 		// Done.
 		waitCh <- struct{}{}
 	}).Once()
+	s.mockEventsCache.On("putEvent", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return().Once()
 
 	// Start timer Processor.
 	emptyResponse := &persistence.GetTimerIndexTasksResponse{Timers: []*persistence.TimerTaskInfo{}}

service/history/transferQueueActiveProcessor.go

@@ -389,12 +389,9 @@ func (t *transferQueueActiveProcessorImpl) processCloseExecution(task *persisten
 
 	executionInfo := msBuilder.GetExecutionInfo()
 	replyToParentWorkflow := msBuilder.HasParentExecution() && executionInfo.CloseStatus != persistence.WorkflowCloseStatusContinuedAsNew
-	var completionEvent *workflow.HistoryEvent
-	if replyToParentWorkflow {
-		completionEvent, ok = msBuilder.GetCompletionEvent()
-		if !ok {
-			return &workflow.InternalServiceError{Message: "Unable to get workflow completion event."}
-		}
+	completionEvent, ok := msBuilder.GetCompletionEvent()
+	if !ok {
+		return &workflow.InternalServiceError{Message: "Unable to get workflow completion event."}
 	}
 	parentDomainID := executionInfo.ParentDomainID
 	parentWorkflowID := executionInfo.ParentWorkflowID
@@ -403,7 +400,7 @@ func (t *transferQueueActiveProcessorImpl) processCloseExecution(task *persisten
 
 	workflowTypeName := executionInfo.WorkflowTypeName
 	workflowStartTimestamp := executionInfo.StartTimestamp.UnixNano()
-	workflowCloseTimestamp := msBuilder.GetLastUpdatedTimestamp()
+	workflowCloseTimestamp := completionEvent.GetTimestamp()
 	workflowCloseStatus := getWorkflowExecutionCloseStatus(executionInfo.CloseStatus)
 	workflowHistoryLength := msBuilder.GetNextEventID() - 1
 

service/history/transferQueueStandbyProcessor.go

@@ -291,10 +291,15 @@ func (t *transferQueueStandbyProcessorImpl) processCloseExecution(transferTask *
 			return nil
 		}
 
+		completionEvent, ok := msBuilder.GetCompletionEvent()
+		if !ok {
+			return &workflow.InternalServiceError{Message: "Unable to get workflow completion event."}
+		}
+
 		executionInfo := msBuilder.GetExecutionInfo()
 		workflowTypeName := executionInfo.WorkflowTypeName
 		workflowStartTimestamp := executionInfo.StartTimestamp.UnixNano()
-		workflowCloseTimestamp := msBuilder.GetLastUpdatedTimestamp()
+		workflowCloseTimestamp := completionEvent.GetTimestamp()
 		workflowCloseStatus := getWorkflowExecutionCloseStatus(executionInfo.CloseStatus)
 		workflowHistoryLength := msBuilder.GetNextEventID() - 1