Try detecting timer and activity resurrection (#4375)

common/dynamicconfig/constants.go

@@ -810,6 +810,13 @@ const (
 	// Default value: 3*time.Minute
 	// Allowed filters: DomainID
 	StandbyTaskReReplicationContextTimeout
+	// ResurrectionCheckMinDelay is the minimal timer processing delay before scanning history to see
+	// if there's a resurrected timer/activity
+	// KeyName: history.resurrectionCheckMinDelay
+	// Value type: Duration
+	// Default value: 24*time.Hour
+	// Allowed filters: N/A
+	ResurrectionCheckMinDelay
 	// QueueProcessorEnableSplit is indicates whether processing queue split policy should be enabled
 	// KeyName: history.queueProcessorEnableSplit
 	// Value type: Bool
@@ -2091,6 +2098,7 @@ var Keys = map[Key]string{
 	StandbyTaskRedispatchInterval:                      "history.standbyTaskRedispatchInterval",
 	TaskRedispatchIntervalJitterCoefficient:            "history.taskRedispatchIntervalJitterCoefficient",
 	StandbyTaskReReplicationContextTimeout:             "history.standbyTaskReReplicationContextTimeout",
+	ResurrectionCheckMinDelay:                          "history.resurrectionCheckMinDelay",
 	QueueProcessorEnableSplit:                          "history.queueProcessorEnableSplit",
 	QueueProcessorSplitMaxLevel:                        "history.queueProcessorSplitMaxLevel",
 	QueueProcessorEnableRandomSplitByDomainID:          "history.queueProcessorEnableRandomSplitByDomainID",

common/metrics/defs.go

@@ -1901,6 +1901,8 @@ const (
 	DecisionAttemptTimer
 	StaleMutableStateCounter
 	DataInconsistentCounter
+	TimerResurrectionCounter
+	ActivityResurrectionCounter
 	AutoResetPointsLimitExceededCounter
 	AutoResetPointCorruptionCounter
 	ConcurrencyUpdateFailureCounter
@@ -2405,6 +2407,8 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		DecisionAttemptTimer:                              {metricName: "decision_attempt", metricType: Timer},
 		StaleMutableStateCounter:                          {metricName: "stale_mutable_state", metricType: Counter},
 		DataInconsistentCounter:                           {metricName: "data_inconsistent", metricType: Counter},
+		TimerResurrectionCounter:                          {metricName: "timer_resurrection", metricType: Counter},
+		ActivityResurrectionCounter:                       {metricName: "activity_resurrection", metricType: Counter},
 		AutoResetPointsLimitExceededCounter:               {metricName: "auto_reset_points_exceed_limit", metricType: Counter},
 		AutoResetPointCorruptionCounter:                   {metricName: "auto_reset_point_corruption", metricType: Counter},
 		ConcurrencyUpdateFailureCounter:                   {metricName: "concurrency_update_failure", metricType: Counter},

service/history/config/config.go

@@ -99,6 +99,7 @@ type Config struct {
 	TaskRedispatchIntervalJitterCoefficient dynamicconfig.FloatPropertyFn
 	StandbyTaskReReplicationContextTimeout  dynamicconfig.DurationPropertyFnWithDomainIDFilter
 	EnableDropStuckTaskByDomainID           dynamicconfig.BoolPropertyFnWithDomainIDFilter
+	ResurrectionCheckMinDelay               dynamicconfig.DurationPropertyFn
 
 	// QueueProcessor settings
 	QueueProcessorEnableSplit                          dynamicconfig.BoolPropertyFn
@@ -378,6 +379,7 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, storeType string, isA
 		TaskRedispatchIntervalJitterCoefficient: dc.GetFloat64Property(dynamicconfig.TaskRedispatchIntervalJitterCoefficient, 0.15),
 		StandbyTaskReReplicationContextTimeout:  dc.GetDurationPropertyFilteredByDomainID(dynamicconfig.StandbyTaskReReplicationContextTimeout, 3*time.Minute),
 		EnableDropStuckTaskByDomainID:           dc.GetBoolPropertyFilteredByDomainID(dynamicconfig.EnableDropStuckTaskByDomainID, false),
+		ResurrectionCheckMinDelay:               dc.GetDurationProperty(dynamicconfig.ResurrectionCheckMinDelay, 24*time.Hour),
 
 		QueueProcessorEnableSplit:                          dc.GetBoolProperty(dynamicconfig.QueueProcessorEnableSplit, false),
 		QueueProcessorSplitMaxLevel:                        dc.GetIntProperty(dynamicconfig.QueueProcessorSplitMaxLevel, 2), // 3 levels, start from 0

service/history/execution/mutable_state.go

@@ -112,6 +112,8 @@ type (
 		CreateNewHistoryEventWithTimestamp(eventType types.EventType, timestamp int64) *types.HistoryEvent
 		CreateTransientDecisionEvents(di *DecisionInfo, identity string) (*types.HistoryEvent, *types.HistoryEvent)
 		DeleteDecision()
+		DeleteUserTimer(timerID string) error
+		DeleteActivity(scheduleEventID int64) error
 		DeleteSignalRequested(requestID string)
 		FailDecision(bool)
 		FlushBufferedEvents() error

service/history/execution/timer_sequence.go

@@ -82,7 +82,7 @@ type (
 
 	// TimerSequence manages user / activity timer
 	TimerSequence interface {
-		IsExpired(referenceTime time.Time, TimerSequenceID TimerSequenceID) bool
+		IsExpired(referenceTime time.Time, TimerSequenceID TimerSequenceID) (time.Duration, bool)
 
 		CreateNextUserTimer() (bool, error)
 		CreateNextActivityTimer() (bool, error)
@@ -113,11 +113,19 @@ func NewTimerSequence(
 func (t *timerSequenceImpl) IsExpired(
 	referenceTime time.Time,
 	TimerSequenceID TimerSequenceID,
-) bool {
+) (time.Duration, bool) {
 
 	// Cassandra timestamp resolution is in millisecond
 	// here we do the check in terms of second resolution.
-	return TimerSequenceID.Timestamp.Unix() <= referenceTime.Unix()
+
+	timerFireTimeInSecond := TimerSequenceID.Timestamp.Unix()
+	referenceTimeInSecond := referenceTime.Unix()
+
+	if timerFireTimeInSecond <= referenceTimeInSecond {
+		return time.Duration(referenceTimeInSecond-timerFireTimeInSecond) * time.Second, true
+	}
+
+	return 0, false
 }
 
 func (t *timerSequenceImpl) CreateNextUserTimer() (bool, error) {