Correctly stop shard on rangeID update failure (#168)

This change whitelists error codes that do not trigger rangeID updates for writes.
If the rangeID update fail, we fallback to immediately stopping the shard. This was done incorrectly before.
Also, add more logging.

Author: Tamer Eldeeb

service/history/historyEngine_test.go

@@ -198,6 +198,7 @@ func (s *engineSuite) TestRespondDecisionTaskCompletedUpdateExecutionFailed() {
 	s.mockExecutionMgr.On("GetWorkflowExecution", mock.Anything).Return(gwmsResponse, nil).Once()
 	s.mockHistoryMgr.On("AppendHistoryEvents", mock.Anything).Return(nil).Once()
 	s.mockExecutionMgr.On("UpdateWorkflowExecution", mock.Anything).Return(errors.New("FAILED")).Once()
+	s.mockShardManager.On("UpdateShard", mock.Anything).Return(nil).Once()
 
 	err := s.mockHistoryEngine.RespondDecisionTaskCompleted(&history.RespondDecisionTaskCompletedRequest{
 		DomainUUID: common.StringPtr(domainID),
@@ -872,6 +873,7 @@ func (s *engineSuite) TestRespondActivityTaskCompletedUpdateExecutionFailed() {
 	s.mockExecutionMgr.On("GetWorkflowExecution", mock.Anything).Return(gwmsResponse, nil).Once()
 	s.mockHistoryMgr.On("AppendHistoryEvents", mock.Anything).Return(nil).Once()
 	s.mockExecutionMgr.On("UpdateWorkflowExecution", mock.Anything).Return(errors.New("FAILED")).Once()
+	s.mockShardManager.On("UpdateShard", mock.Anything).Return(nil).Once()
 
 	err := s.mockHistoryEngine.RespondActivityTaskCompleted(&history.RespondActivityTaskCompletedRequest{
 		DomainUUID: common.StringPtr(domainID),
@@ -1246,6 +1248,7 @@ func (s *engineSuite) TestRespondActivityTaskFailedUpdateExecutionFailed() {
 	s.mockExecutionMgr.On("GetWorkflowExecution", mock.Anything).Return(gwmsResponse, nil).Once()
 	s.mockHistoryMgr.On("AppendHistoryEvents", mock.Anything).Return(nil).Once()
 	s.mockExecutionMgr.On("UpdateWorkflowExecution", mock.Anything).Return(errors.New("FAILED")).Once()
+	s.mockShardManager.On("UpdateShard", mock.Anything).Return(nil).Once()
 
 	err := s.mockHistoryEngine.RespondActivityTaskFailed(&history.RespondActivityTaskFailedRequest{
 		DomainUUID: common.StringPtr(domainID),

service/history/loggingHelpers.go

@@ -104,6 +104,7 @@ const (
 	tagValueStoreOperationGetWorkflowExecution    = "get-wf-execution"
 	tagValueStoreOperationUpdateWorkflowExecution = "update-wf-execution"
 	tagValueStoreOperationDeleteWorkflowExecution = "delete-wf-execution"
+	tagValueStoreOperationUpdateShard             = "update-shard"
 )
 
 func logInvalidHistoryActionEvent(logger bark.Logger, action string, eventID int64, state string) {

service/history/shardContext.go

@@ -1,6 +1,7 @@
 package history
 
 import (
+	"fmt"
 	"sync"
 	"sync/atomic"
 
@@ -43,7 +44,7 @@ type (
 		timerSequenceNumber int64
 		rangeSize           uint
 		closeCh             chan<- int
-		isClosed            int32
+		isClosed            bool
 		logger              bark.Logger
 		metricsClient       metrics.Client
 
@@ -132,6 +133,7 @@ func (s *shardContextImpl) CreateWorkflowExecution(request *persistence.CreateWo
 		if err != nil {
 			return nil, err
 		}
+		s.logger.Debugf("Assigning transfer task ID: %v", id)
 		task.SetTaskID(id)
 		transferMaxReadLevel = id
 	}
@@ -155,7 +157,8 @@ Create_Loop:
 						s.closeShard()
 					}
 				}
-			case *shared.InternalServiceError, *persistence.TimeoutError:
+			case *shared.WorkflowExecutionAlreadyStartedError:
+			default:
 				{
 					// We have no idea if the write failed or will eventually make it to
 					// persistence. Increment RangeID to guarantee that subsequent reads
@@ -191,6 +194,7 @@ func (s *shardContextImpl) UpdateWorkflowExecution(request *persistence.UpdateWo
 		if err != nil {
 			return err
 		}
+		s.logger.Debugf("Assigning transfer task ID: %v", id)
 		task.SetTaskID(id)
 		transferMaxReadLevel = id
 	}
@@ -201,6 +205,7 @@ func (s *shardContextImpl) UpdateWorkflowExecution(request *persistence.UpdateWo
 			if err != nil {
 				return err
 			}
+			s.logger.Debugf("Assigning transfer task ID: %v", id)
 			task.SetTaskID(id)
 			transferMaxReadLevel = id
 		}
@@ -225,7 +230,8 @@ Update_Loop:
 						s.closeShard()
 					}
 				}
-			case *shared.InternalServiceError, *persistence.TimeoutError:
+			case *persistence.ConditionFailedError:
+			default:
 				{
 					// We have no idea if the write failed or will eventually make it to
 					// persistence. Increment RangeID to guarantee that subsequent reads
@@ -277,11 +283,15 @@ func (s *shardContextImpl) getRangeID() int64 {
 }
 
 func (s *shardContextImpl) closeShard() {
-	if !atomic.CompareAndSwapInt32(&s.isClosed, 0, 1) {
+	if s.isClosed {
 		return
 	}
 
-	s.rangeID = -1 // fails any writes that may start after this point.
+	s.isClosed = true
+
+	// fails any writes that may start after this point.
+	s.shardInfo.RangeID = -1
+	atomic.StoreInt64(&s.rangeID, s.shardInfo.RangeID)
 
 	if s.closeCh != nil {
 		// This is the channel passed in by shard controller to monitor if a shard needs to be unloaded
@@ -320,6 +330,8 @@ func (s *shardContextImpl) renewRangeLocked(isStealing bool) error {
 		ShardInfo:       updatedShardInfo,
 		PreviousRangeID: s.shardInfo.RangeID})
 	if err != nil {
+		logPersistantStoreErrorEvent(s.logger, tagValueStoreOperationUpdateShard, err,
+			fmt.Sprintf("{RangeID: %v}", s.shardInfo.RangeID))
 		// Shard is stolen, trigger history engine shutdown
 		if _, ok := err.(*persistence.ShardOwnershipLostError); ok {
 			s.closeShard()