Auto-Reset: Skip reset task if current run has changed (cadence-workf…

…low#1777)

common/persistence/cassandra/cassandraPersistence.go

@@ -3058,10 +3058,9 @@ func (d *cassandraPersistence) createTransferTasks(batch *gocql.Batch, transferT
 			targetWorkflowID = task.(*p.StartChildExecutionTask).TargetWorkflowID
 			scheduleID = task.(*p.StartChildExecutionTask).InitiatedID
 
-		case p.TransferTaskTypeCloseExecution:
-			// No explicit property needs to be set
-
-		case p.TransferTaskTypeRecordWorkflowStarted:
+		case p.TransferTaskTypeCloseExecution,
+			p.TransferTaskTypeRecordWorkflowStarted,
+			p.TransferTaskTypeResetWorkflow:
 			// No explicit property needs to be set
 
 		default:

common/persistence/sql/sqlExecutionManager.go

@@ -1614,10 +1614,9 @@ func createTransferTasks(tx sqldb.Tx, transferTasks []p.Task, shardID int, domai
 			info.TargetWorkflowID = &task.(*p.StartChildExecutionTask).TargetWorkflowID
 			info.ScheduleID = &task.(*p.StartChildExecutionTask).InitiatedID
 
-		case p.TransferTaskTypeCloseExecution:
-			// No explicit property needs to be set
-
-		case p.TransferTaskTypeRecordWorkflowStarted:
+		case p.TransferTaskTypeCloseExecution,
+			p.TransferTaskTypeRecordWorkflowStarted,
+			p.TransferTaskTypeResetWorkflow:
 			// No explicit property needs to be set
 
 		default:

service/history/historyEngine2_test.go

@@ -116,7 +116,7 @@ func (s *engine2Suite) SetupTest() {
 	s.mockClusterMetadata.On("GetAllClusterFailoverVersions").Return(cluster.TestSingleDCAllClusterFailoverVersions)
 	s.mockClusterMetadata.On("IsGlobalDomainEnabled").Return(false)
 	s.mockDomainCache = &cache.DomainCacheMock{}
-	s.mockDomainCache.On("GetDomainByID", mock.Anything).Return(cache.NewDomainCacheEntryForTest(&p.DomainInfo{ID: validDomainID}, nil), nil)
+	s.mockDomainCache.On("GetDomainByID", mock.Anything).Return(cache.NewDomainCacheEntryForTest(&p.DomainInfo{ID: validDomainID}, &p.DomainConfig{}), nil)
 	s.mockEventsCache = &MockEventsCache{}
 	s.mockEventsCache.On("putEvent", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
 		mock.Anything).Return()

service/history/transferQueueActiveProcessor.go

@@ -914,6 +914,21 @@ func (t *transferQueueActiveProcessorImpl) processResetWorkflow(task *persistenc
 		logger.Warn("Auto-Reset is skipped, because current run is deleted.")
 		return nil
 	}
+	if !currMutableState.IsWorkflowExecutionRunning() {
+		//it means this this might not be current anymore, we need to check
+		var resp *persistence.GetCurrentExecutionResponse
+		resp, retError = t.executionManager.GetCurrentExecution(&persistence.GetCurrentExecutionRequest{
+			DomainID:   task.DomainID,
+			WorkflowID: task.WorkflowID,
+		})
+		if retError != nil {
+			return
+		}
+		if resp.RunID != task.RunID {
+			logger.Warn("Auto-Reset is skipped, because current run is stale.")
+			return nil
+		}
+	}
 	// TODO: current reset doesn't allow childWFs, in the future we will release this restriction
 	if len(currMutableState.GetPendingChildExecutionInfos()) > 0 {
 		logger.Warn("Auto-Reset is skipped, because current run has pending child executions.")

service/history/workflowExecutionContext.go

@@ -386,7 +386,7 @@ func (c *workflowExecutionContextImpl) updateWorkflowExecutionForActive(transfer
 	}
 
 	// compare with bad binaries and schedule a reset task
-	if len(c.msBuilder.GetPendingChildExecutionInfos()) > 0 {
+	if len(c.msBuilder.GetPendingChildExecutionInfos()) == 0 {
 		// only schedule reset task if current doesn't have childWFs.
 		// TODO: This will be removed once our reset allows childWFs