add TimeSkippingInfoUtil and unexpected data loss error

Author: feiyang3cat

service/history/timer_queue_active_task_executor.go

@@ -2,7 +2,6 @@ package history
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"time"
 
@@ -26,6 +25,7 @@ import (
 	"go.temporal.io/server/common/primitives/timestamp"
 	"go.temporal.io/server/common/priorities"
 	"go.temporal.io/server/common/resource"
+	"go.temporal.io/server/common/softassert"
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/consts"
 	"go.temporal.io/server/service/history/deletemanager"
@@ -947,10 +947,8 @@ func (t *timerQueueActiveTaskExecutor) executeTimeSkippingTimerTask(
 		release(nil)
 		return errNoTimerFired
 	}
-	// 2) fast-forward is disabled/reached, or this task was superseded by a newer fast-forward.
-	// The versioned transition's TransitionCount identifies the specific fast-forward instance
-	// (a re-applied fast-forward records a fresh transition), so a mismatch means this task is
-	// stale and is dropped silently.
+
+	// 2) match current pending fast-forward
 	if !tsiUtil.HasPendingFastForward() {
 		release(nil)
 		return errNoTimerFired
@@ -965,11 +963,17 @@ func (t *timerQueueActiveTaskExecutor) executeTimeSkippingTimerTask(
 		taskVersion := task.VersionedTransition.GetNamespaceFailoverVersion()
 		ffVersion := ffVT.GetNamespaceFailoverVersion()
 		if err := CheckTaskVersion(t.shardContext, t.logger, mutableState.GetNamespaceEntry(), ffVersion, taskVersion, task); err != nil {
-			release(nil)
-			return errNoTimerFired
+			// ErrTaskVersionMismatch: the framework drops the task and records the TaskVersionMisMatch metric.
+			return err
 		}
 	} else {
-		return errors.New("time skipping timer task validation failed: nil versioned transition")
+		// Invariant: for a pending fast-forward and a task both exist, they must both have a non-nil versioned transition.
+		errorMsg := fmt.Sprintf(
+			"time-skipping timer task validation failed: VersionedTransition is nil (task: %t, pending fast-forward: %t)",
+			task.VersionedTransition == nil,
+			ffVT == nil,
+		)
+		return softassert.UnexpectedDataLoss(t.logger, errorMsg, nil)
 	}
 
 	// 3) firing fast-forward timer (only turns off time skipping, and no task regeneration)

service/history/workflow/timeskipping_test.go

@@ -1306,3 +1306,76 @@ func (s *mutableStateSuite) TestWrapTimeSourceWithTimeSkipping() {
 		s.Equal(fixedBase.Add(skipped), event.GetEventTime().AsTime())
 	})
 }
+
+func (s *mutableStateSuite) TestTimeSkippingInfoUtil() {
+	s.Run("NilTSISafe", func() {
+		util := NewTimeSkippingInfoUtil(nil)
+		s.False(util.HasPendingFastForward())
+		s.False(util.IsEnabled())
+	})
+
+	s.Run("NilFFISafe", func() {
+		s.mutableState.executionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
+			Config:          &commonpb.TimeSkippingConfig{Enabled: true},
+			FastForwardInfo: nil,
+		}
+		util := NewTimeSkippingInfoUtil(s.mutableState.executionInfo.TimeSkippingInfo)
+		s.False(util.HasPendingFastForward())
+	})
+
+	s.Run("NoPendingFastForwardWhenHasReached", func() {
+		s.mutableState.executionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
+			Config: &commonpb.TimeSkippingConfig{Enabled: true},
+			FastForwardInfo: &persistencespb.FastForwardInfo{
+				HasReached: true,
+				TargetTime: timestamppb.New(time.Now()),
+				LastUpdateVersionedTransition: &persistencespb.VersionedTransition{
+					TransitionCount:          1,
+					NamespaceFailoverVersion: 1,
+				},
+			},
+		}
+		util := NewTimeSkippingInfoUtil(s.mutableState.executionInfo.TimeSkippingInfo)
+		s.False(util.HasPendingFastForward())
+	})
+
+	s.Run("NoPendingFastForwardWhenDisabled", func() {
+		s.mutableState.executionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
+			Config: &commonpb.TimeSkippingConfig{Enabled: false},
+			FastForwardInfo: &persistencespb.FastForwardInfo{
+				HasReached: false,
+				TargetTime: timestamppb.New(time.Now()),
+				LastUpdateVersionedTransition: &persistencespb.VersionedTransition{
+					TransitionCount:          1,
+					NamespaceFailoverVersion: 1,
+				},
+			},
+		}
+		util := NewTimeSkippingInfoUtil(s.mutableState.executionInfo.TimeSkippingInfo)
+		s.False(util.HasPendingFastForward())
+	})
+
+	s.Run("HasPendingFastForward", func() {
+		s.mutableState.executionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
+			Config: &commonpb.TimeSkippingConfig{Enabled: true},
+			FastForwardInfo: &persistencespb.FastForwardInfo{
+				HasReached: false,
+				TargetTime: timestamppb.New(time.Now()),
+				LastUpdateVersionedTransition: &persistencespb.VersionedTransition{
+					TransitionCount:          1,
+					NamespaceFailoverVersion: 1,
+				},
+			},
+		}
+		util := NewTimeSkippingInfoUtil(s.mutableState.executionInfo.TimeSkippingInfo)
+		s.True(util.HasPendingFastForward())
+	})
+
+	s.Run("IsEnabled", func() {
+		s.mutableState.executionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
+			Config: &commonpb.TimeSkippingConfig{Enabled: true},
+		}
+		util := NewTimeSkippingInfoUtil(s.mutableState.executionInfo.TimeSkippingInfo)
+		s.True(util.IsEnabled())
+	})
+}