Bugfix: failover trigger should also notify activer timer / transfer …

common/cache/domainCache.go

@@ -67,7 +67,7 @@ type (
 	// requests using the stale entry from cache upto an hour
 	DomainCache interface {
 		common.Daemon
-		RegisterDomainChangeCallback(shard int, initialNotificationVersion int64, fn CallbackFn)
+		RegisterDomainChangeCallback(shard int, initialNotificationVersion int64, beforeCallback CallbackFn, afterCallback CallbackFn)
 		UnregisterDomainChangeCallback(shard int)
 		GetDomain(name string) (*DomainCacheEntry, error)
 		GetDomainByID(id string) (*DomainCacheEntry, error)
@@ -89,7 +89,8 @@ type (
 
 		sync.RWMutex
 		domainNotificationVersion int64
-		callbacks                 map[int]CallbackFn
+		beforeCallbacks           map[int]CallbackFn
+		afterCallbacks            map[int]CallbackFn
 	}
 
 	// DomainCacheEntries is DomainCacheEntry slice
@@ -126,7 +127,8 @@ func NewDomainCache(metadataMgr persistence.MetadataManager, clusterMetadata clu
 		clusterMetadata: clusterMetadata,
 		timeSource:      common.NewRealTimeSource(),
 		logger:          logger,
-		callbacks:       make(map[int]CallbackFn),
+		beforeCallbacks: make(map[int]CallbackFn),
+		afterCallbacks:  make(map[int]CallbackFn),
 	}
 }
 
@@ -178,14 +180,17 @@ func (c *domainCache) GetAllDomain() map[string]*DomainCacheEntry {
 }
 
 // RegisterDomainChangeCallback set a domain change callback
-// WARN: the callback function will be triggered by domain cache when holding the domain cache lock,
+// WARN: the beforeCallback function will be triggered by domain cache when holding the domain cache lock,
 // make sure the callback function will not call domain cache again in case of dead lock
-func (c *domainCache) RegisterDomainChangeCallback(shard int, initialNotificationVersion int64, fn CallbackFn) {
+// afterCallback will be invoked when NOT holding the domain cache lock.
+func (c *domainCache) RegisterDomainChangeCallback(shard int, initialNotificationVersion int64, beforeCallback CallbackFn, afterCallback CallbackFn) {
 	c.Lock()
 	defer c.Unlock()
 
-	c.callbacks[shard] = fn
+	c.beforeCallbacks[shard] = beforeCallback
+	c.afterCallbacks[shard] = afterCallback
 
+	// this section is trying to make the shard catch up with domain changes
 	if c.domainNotificationVersion > initialNotificationVersion {
 		domains := DomainCacheEntries{}
 		for _, domain := range c.GetAllDomain() {
@@ -197,7 +202,8 @@ func (c *domainCache) RegisterDomainChangeCallback(shard int, initialNotificatio
 		sort.Sort(domains)
 		for _, domain := range domains {
 			if domain.notificationVersion >= initialNotificationVersion {
-				fn(nil, domain)
+				beforeCallback(nil, domain)
+				afterCallback(nil, domain)
 			}
 		}
 	}
@@ -208,7 +214,8 @@ func (c *domainCache) UnregisterDomainChangeCallback(shard int) {
 	c.Lock()
 	defer c.Unlock()
 
-	delete(c.callbacks, shard)
+	delete(c.beforeCallbacks, shard)
+	delete(c.afterCallbacks, shard)
 }
 
 // GetDomain retrieves the information from the cache if it exists, otherwise retrieves the information from metadata
@@ -343,9 +350,8 @@ func (c *domainCache) updateIDToDomainCache(id string, record *DomainCacheEntry)
 		return nil, err
 	}
 	entry := elem.(*DomainCacheEntry)
-	entry.Lock()
-	defer entry.Unlock()
 
+	entry.Lock()
 	var prevDomain *DomainCacheEntry
 	triggerCallback := c.clusterMetadata.IsGlobalDomainEnabled() &&
 		// expiry will be non zero when the entry is initialized / valid
@@ -368,7 +374,11 @@ func (c *domainCache) updateIDToDomainCache(id string, record *DomainCacheEntry)
 
 	nextDomain := entry.duplicate()
 	if triggerCallback {
-		c.triggerDomainChangeCallback(prevDomain, nextDomain)
+		c.triggerDomainBeforeChangeCallback(prevDomain, nextDomain)
+	}
+	entry.Unlock()
+	if triggerCallback {
+		c.triggerDomainAfterChangeCallback(prevDomain, nextDomain)
 	}
 
 	return nextDomain, nil
@@ -445,10 +455,18 @@ func (c *domainCache) getDomainByID(id string) (*DomainCacheEntry, error) {
 	return newEntry, nil
 }
 
-func (c *domainCache) triggerDomainChangeCallback(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
+func (c *domainCache) triggerDomainBeforeChangeCallback(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
+	c.RLock()
+	defer c.RUnlock()
+	for _, callback := range c.beforeCallbacks {
+		callback(prevDomain, nextDomain)
+	}
+}
+
+func (c *domainCache) triggerDomainAfterChangeCallback(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
 	c.RLock()
 	defer c.RUnlock()
-	for _, callback := range c.callbacks {
+	for _, callback := range c.afterCallbacks {
 		callback(prevDomain, nextDomain)
 	}
 }

common/cache/domainCache_mock.go

@@ -145,9 +145,9 @@ func (_m *DomainCacheMock) GetDomainNotificationVersion() int64 {
 	return r0
 }
 
-// RegisterDomainChangeCallback provides a mock function with given fields: shard, initialNotificationVersion, fn
-func (_m *DomainCacheMock) RegisterDomainChangeCallback(shard int, initialNotificationVersion int64, fn CallbackFn) {
-	_m.Called(shard, initialNotificationVersion, fn)
+// RegisterDomainChangeCallback provides a mock function with given fields: shard, initialNotificationVersion, beforeCallback, afterCallback
+func (_m *DomainCacheMock) RegisterDomainChangeCallback(shard int, initialNotificationVersion int64, beforeCallback CallbackFn, afterCallback CallbackFn) {
+	_m.Called(shard, initialNotificationVersion, beforeCallback, afterCallback)
 }
 
 // Start provides a mock function with given fields:

common/cache/domainCache_test.go

@@ -277,16 +277,26 @@ func (s *domainCacheSuite) TestRegisterCallback_CatchUp() {
 	s.Nil(s.domainCache.refreshDomains())
 	s.Equal(domainNotificationVersion, s.domainCache.GetDomainNotificationVersion())
 
-	entriesNotification := []*DomainCacheEntry{}
+	entriesNotificationBefore := []*DomainCacheEntry{}
+	entriesNotificationAfter := []*DomainCacheEntry{}
 	// we are not testing catching up, so make this really large
 	currentDomainNotificationVersion := int64(0)
-	s.domainCache.RegisterDomainChangeCallback(0, currentDomainNotificationVersion, func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
-		s.Nil(prevDomain)
-		entriesNotification = append(entriesNotification, nextDomain)
-	})
+	s.domainCache.RegisterDomainChangeCallback(
+		0,
+		currentDomainNotificationVersion,
+		func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
+			s.Nil(prevDomain)
+			entriesNotificationBefore = append(entriesNotificationBefore, nextDomain)
+		},
+		func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
+			s.Nil(prevDomain)
+			entriesNotificationAfter = append(entriesNotificationAfter, nextDomain)
+		},
+	)
 
 	// the order matters here, should be ordered by notification version
-	s.Equal([]*DomainCacheEntry{entry1, entry2}, entriesNotification)
+	s.Equal([]*DomainCacheEntry{entry1, entry2}, entriesNotificationBefore)
+	s.Equal([]*DomainCacheEntry{entry1, entry2}, entriesNotificationAfter)
 }
 
 func (s *domainCacheSuite) TestUpdateCache_ListTrigger() {
@@ -377,16 +387,28 @@ func (s *domainCacheSuite) TestUpdateCache_ListTrigger() {
 	entry1New := s.buildEntryFromRecord(domainRecord1New)
 	domainNotificationVersion++
 
-	entriesOld := []*DomainCacheEntry{}
-	entriesNew := []*DomainCacheEntry{}
+	entriesOldBefore := []*DomainCacheEntry{}
+	entriesNewBefore := []*DomainCacheEntry{}
+	entriesOldAfter := []*DomainCacheEntry{}
+	entriesNewAfter := []*DomainCacheEntry{}
 	// we are not testing catching up, so make this really large
 	currentDomainNotificationVersion := int64(9999999)
-	s.domainCache.RegisterDomainChangeCallback(0, currentDomainNotificationVersion, func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
-		entriesOld = append(entriesOld, prevDomain)
-		entriesNew = append(entriesNew, nextDomain)
-	})
-	s.Empty(entriesOld)
-	s.Empty(entriesNew)
+	s.domainCache.RegisterDomainChangeCallback(
+		0,
+		currentDomainNotificationVersion,
+		func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
+			entriesOldBefore = append(entriesOldBefore, prevDomain)
+			entriesNewBefore = append(entriesNewBefore, nextDomain)
+		},
+		func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
+			entriesOldAfter = append(entriesOldAfter, prevDomain)
+			entriesNewAfter = append(entriesNewAfter, nextDomain)
+		},
+	)
+	s.Empty(entriesOldBefore)
+	s.Empty(entriesNewBefore)
+	s.Empty(entriesOldAfter)
+	s.Empty(entriesNewAfter)
 
 	s.metadataMgr.On("GetMetadata").Return(&persistence.GetMetadataResponse{NotificationVersion: domainNotificationVersion}, nil).Once()
 	s.metadataMgr.On("ListDomains", &persistence.ListDomainsRequest{
@@ -402,8 +424,10 @@ func (s *domainCacheSuite) TestUpdateCache_ListTrigger() {
 	// the record 1 got updated later, thus a higher notification version.
 	// making sure notifying from lower to higher version helps the shard to keep track the
 	// domain change events
-	s.Equal([]*DomainCacheEntry{entry2Old, entry1Old}, entriesOld)
-	s.Equal([]*DomainCacheEntry{entry2New, entry1New}, entriesNew)
+	s.Equal([]*DomainCacheEntry{entry2Old, entry1Old}, entriesOldBefore)
+	s.Equal([]*DomainCacheEntry{entry2New, entry1New}, entriesNewBefore)
+	s.Equal([]*DomainCacheEntry{entry2Old, entry1Old}, entriesOldAfter)
+	s.Equal([]*DomainCacheEntry{entry2New, entry1New}, entriesNewAfter)
 }
 
 func (s *domainCacheSuite) TestUpdateCache_GetNotTrigger() {
@@ -442,19 +466,30 @@ func (s *domainCacheSuite) TestUpdateCache_GetNotTrigger() {
 	s.Nil(err)
 	s.Equal(entryOld, entry)
 
-	callbackInvoked := false
+	callbackBeforeInvoked := false
+	callbackAfterInvoked := false
 	// we are not testing catching up, so make this really large
 	currentDomainNotificationVersion := int64(9999999)
-	s.domainCache.RegisterDomainChangeCallback(0, currentDomainNotificationVersion, func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
-		s.Equal(entryOld, prevDomain)
-		s.Equal(entryNew, nextDomain)
-		callbackInvoked = true
-	})
+	s.domainCache.RegisterDomainChangeCallback(
+		0,
+		currentDomainNotificationVersion,
+		func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
+			s.Equal(entryOld, prevDomain)
+			s.Equal(entryNew, nextDomain)
+			callbackBeforeInvoked = true
+		},
+		func(prevDomain *DomainCacheEntry, nextDomain *DomainCacheEntry) {
+			s.Equal(entryOld, prevDomain)
+			s.Equal(entryNew, nextDomain)
+			callbackAfterInvoked = true
+		},
+	)
 
 	entry, err = s.domainCache.updateIDToDomainCache(domainRecordNew.Info.ID, entryNew)
 	s.Nil(err)
 	s.Equal(entryNew, entry)
-	s.False(callbackInvoked)
+	s.False(callbackBeforeInvoked)
+	s.False(callbackAfterInvoked)
 }
 
 func (s *domainCacheSuite) TestGetUpdateCache_ConcurrentAccess() {

common/metrics/defs.go

@@ -708,6 +708,7 @@ const (
 	ReplicatorMessages = iota + NumCommonMetrics
 	ReplicatorFailures
 	ReplicatorLatency
+	ReplicatorRetryPercentage
 )
 
 // MetricDefs record the metrics for all services
@@ -801,9 +802,10 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		BufferThrottleCounter:         {metricName: "buffer.throttle.count"},
 	},
 	Worker: {
-		ReplicatorMessages: {metricName: "replicator.messages"},
-		ReplicatorFailures: {metricName: "replicator.errors"},
-		ReplicatorLatency:  {metricName: "replicator.latency"},
+		ReplicatorMessages:        {metricName: "replicator.messages"},
+		ReplicatorFailures:        {metricName: "replicator.errors"},
+		ReplicatorLatency:         {metricName: "replicator.latency"},
+		ReplicatorRetryPercentage: {metricName: "replicator.retry-percentage", metricType: Gauge},
 	},
 }
 

service/history/historyEngine.go

@@ -194,26 +194,52 @@ func (e *historyEngineImpl) Stop() {
 }
 
 func (e *historyEngineImpl) registerDomainFailoverCallback() {
+
+	failoverPredicate := func(nextDomain *cache.DomainCacheEntry, action func()) {
+		domainFailoverNotificationVersion := nextDomain.GetFailoverNotificationVersion()
+		shardNotificationVersion := e.shard.GetDomainNotificationVersion()
+		domainActiveCluster := nextDomain.GetReplicationConfig().ActiveClusterName
+
+		if nextDomain.IsGlobalDomain() &&
+			domainFailoverNotificationVersion >= shardNotificationVersion &&
+			domainActiveCluster == e.currentClusterName {
+			action()
+		}
+	}
+
 	// first set the failover callback
 	e.shard.GetDomainCache().RegisterDomainChangeCallback(
 		e.shard.GetShardID(),
 		e.shard.GetDomainCache().GetDomainNotificationVersion(),
+		// before the domain change, this will be invoked when (most of time) domain cache is locked
 		func(prevDomain *cache.DomainCacheEntry, nextDomain *cache.DomainCacheEntry) {
-			domainFailoverNotificationVersion := nextDomain.GetFailoverNotificationVersion()
-			shardNotificationVersion := e.shard.GetDomainNotificationVersion()
-			domainActiveCluster := nextDomain.GetReplicationConfig().ActiveClusterName
-
 			e.logger.Infof("Domain Change Event: Shard: %v, Domain: %v, ID: %v, Failover Notification Version: %v, Active Cluster: %v, Shard Domain Notification Version: %v\n",
-				e.shard.GetShardID(), nextDomain.GetInfo().Name, nextDomain.GetInfo().ID, domainFailoverNotificationVersion, domainActiveCluster, shardNotificationVersion)
+				e.shard.GetShardID(), nextDomain.GetInfo().Name, nextDomain.GetInfo().ID,
+				nextDomain.GetFailoverNotificationVersion(), nextDomain.GetReplicationConfig().ActiveClusterName, e.shard.GetDomainNotificationVersion())
 
-			if nextDomain.IsGlobalDomain() &&
-				domainFailoverNotificationVersion >= shardNotificationVersion &&
-				domainActiveCluster == e.currentClusterName {
-				domainID := prevDomain.GetInfo().ID
+			failoverPredicate(nextDomain, func() {
+				e.logger.Infof("Domain Failover Start: Shard: %v, Domain: %v, ID: %v\n",
+					e.shard.GetShardID(), nextDomain.GetInfo().Name, nextDomain.GetInfo().ID)
+
+				domainID := nextDomain.GetInfo().ID
 				e.txProcessor.FailoverDomain(domainID)
 				e.timerProcessor.FailoverDomain(domainID)
-			}
-
+			})
+		},
+		// after the domain change, this will be invoked when domain cache is NOT locked
+		func(prevDomain *cache.DomainCacheEntry, nextDomain *cache.DomainCacheEntry) {
+			failoverPredicate(nextDomain, func() {
+				e.logger.Infof("Domain Failover Notify Active: Shard: %v, Domain: %v, ID: %v\n",
+					e.shard.GetShardID(), nextDomain.GetInfo().Name, nextDomain.GetInfo().ID)
+
+				now := e.shard.GetTimeSource().Now()
+				// the fake tasks will not be actually used, we just need to make sure
+				// its length > 0 and has correct timestamp, to trkgger a db scan
+				fakeDecisionTask := []persistence.Task{&persistence.DecisionTask{}}
+				fakeDecisionTimeoutTask := []persistence.Task{&persistence.DecisionTimeoutTask{VisibilityTimestamp: now}}
+				e.txProcessor.NotifyNewTask(e.currentClusterName, now, fakeDecisionTask)
+				e.timerProcessor.NotifyNewTimers(e.currentClusterName, now, fakeDecisionTimeoutTask)
+			})
 			e.shard.UpdateDomainNotificationVersion(nextDomain.GetNotificationVersion() + 1)
 		},
 	)

service/worker/processor.go

@@ -193,6 +193,8 @@ func (p *replicationTaskProcessor) getRemainingRetryCount(remainingRetryCount in
 	numWorker := float64(p.config.ReplicatorConcurrency)
 	retryPercentage := workerInRetry / numWorker
 
+	p.metricsClient.UpdateGauge(metrics.ReplicatorScope, metrics.ReplicatorRetryPercentage, retryPercentage)
+
 	min := func(i int64, j int64) int64 {
 		if i < j {
 			return i