Check max taskID instead of max read level when process task (#2371)

temporalio · Jan 12, 2022 · ba30aa8 · ba30aa8
1 parent 00a49ab
commit ba30aa8
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 2 deletions.
diff --git a/service/history/shard/context.go b/service/history/shard/context.go
@@ -69,6 +69,8 @@ type (
 		GetTransferMaxReadLevel() int64
 		UpdateTimerMaxReadLevel(cluster string) time.Time
 
+		GetMaxTaskIDForCurrentRangeID() int64
+
 		SetCurrentTime(cluster string, currentTime time.Time)
 		GetCurrentTime(cluster string) time.Time
 		GetLastUpdatedTime() time.Time

diff --git a/service/history/shard/context_impl.go b/service/history/shard/context_impl.go
@@ -185,6 +185,13 @@ func (s *ContextImpl) GetEngine() (Engine, error) {
 	return s.engine, nil
 }
 
+func (s *ContextImpl) GetMaxTaskIDForCurrentRangeID() int64 {
+	s.rLock()
+	defer s.rUnlock()
+	// maxTransferSequenceNumber is the exclusive upper bound of task ID for current range.
+	return s.maxTransferSequenceNumber - 1
+}
+
 func (s *ContextImpl) GenerateTransferTaskID() (int64, error) {
 	s.wLock()
 	defer s.wUnlock()

diff --git a/service/history/shard/context_mock.go b/service/history/shard/context_mock.go
diff --git a/service/history/taskProcessor.go b/service/history/taskProcessor.go
@@ -168,8 +168,8 @@ func (t *taskProcessor) taskWorker(
 			if !ok {
 				return
 			}
-			if task.GetTaskID() > t.shard.GetTransferMaxReadLevel() {
-				// this could happen if we lost ownership and was not aware of it.
+			if task.GetTaskID() > t.shard.GetMaxTaskIDForCurrentRangeID() {
+				// this could happen if we lost ownership and were not aware of it.
 				// unload shard
 				t.shard.Unload()
 				return