Skip reactivation signals for current/ramping/draining versions

common/dynamicconfig/constants.go

@@ -2921,17 +2921,12 @@ instead of the previous HSM backed implementation.`,
 		`Maximum number of entries in the version membership cache.`,
 	)
 
-	VersionReactivationSignalCacheTTL = NewGlobalDurationSetting(
-		"history.versionReactivationSignalCacheTTL",
-		10*time.Second,
-		`TTL for caching drainage reactivation signals to version workflows. These signals are sent from the history service to update the version workflow's 
-		draining status to DRAINING from DRAINED/INACTIVE states.`,
-	)
-
-	VersionReactivationSignalCacheMaxSize = NewGlobalIntSetting(
-		"history.versionReactivationSignalCacheMaxSize",
+	ReactivationSignalDedupCacheMaxSize = NewGlobalIntSetting(
+		"worker.reactivationSignalDedupCacheMaxSize",
 		10000,
-		`Maximum number of entries in the version reactivation signal cache.`,
+		`Maximum number of entries in the per-pod reactivation-signal dedup cache on the
+		worker deployment client. Each entry tracks the highest revision signaled for one
+		target version workflow.`,
 	)
 
 	EnableVersionReactivationSignals = NewGlobalBoolSetting(

common/metrics/metric_defs.go

@@ -47,7 +47,7 @@ const (
 	MutableStateCacheTypeTagValue                     = "mutablestate"
 	EventsCacheTypeTagValue                           = "events"
 	VersionMembershipCacheTypeTagValue                = "version_membership"
-	VersionReactivationSignalCacheTypeTagValue        = "version_reactivation_signal"
+	ReactivationSignalDedupCacheTypeTagValue          = "reactivation_signal_dedup"
 	RoutingInfoCacheTypeTagValue                      = "routing_info"
 	NexusEndpointRegistryReadThroughCacheTypeTagValue = "nexus_endpoint_registry_readthrough"
 
@@ -459,8 +459,9 @@ const (
 	VersionMembershipCacheGetScope = "VersionMembershipCacheGet"
 	// VersionMembershipCachePutScope is the scope used by version membership cache
 	VersionMembershipCachePutScope = "VersionMembershipCachePut"
-	// VersionReactivationSignalCacheShouldSendScope is the scope used by version reactivation signal cache
-	VersionReactivationSignalCacheShouldSendScope = "VersionReactivationSignalCacheShouldSend"
+	// ReactivationSignalDedupScope is the scope used by the per-pod reactivation-signal
+	// dedup cache on the worker-deployment client.
+	ReactivationSignalDedupScope = "ReactivationSignalDedup"
 	// RoutingInfoCacheGetScope is the scope used by routing info cache
 	RoutingInfoCacheGetScope = "RoutingInfoCacheGet"
 	// RoutingInfoCachePutScope is the scope used by routing info cache

common/worker_versioning/version_membership_cache.go

@@ -7,20 +7,27 @@ import (
 	"go.temporal.io/server/common/metrics"
 )
 
-// VersionMembershipCache is used to cache results of Matching's CheckTaskQueueVersionMembership
-// calls (used internally by the worker versioning pinned override validation).
+// VersionMembershipAndReactivationStatusCache caches results of Matching's
+// CheckTaskQueueVersionMembership calls. It stores three pieces of information per version:
+//   - isMember: whether the task queue exists in the version (used for pinned override validation).
+//   - shouldSkipReactivation: true when the version's current status is CURRENT, RAMPING,
+//     or DRAINING, in which case the caller skips the reactivation signal. The zero value
+//     (false) is the safe default and covers unknown / not-found / old-matching cases.
+//   - revisionNumber: the version's current revision per matching's view. Used as part of the
+//     reactivation signal's RequestId so that all history pods targeting the same version at
+//     the same revision compose the same dedup key. Zero means unknown (old matching server or
+//     legacy DeploymentVersionData format with no revision_number field).
 //
 // Implementations are expected to be safe for concurrent use.
 type (
-	VersionMembershipCache interface {
-		// Get returns (isMember, ok). ok=false means there was no cached value.
+	VersionMembershipAndReactivationStatusCache interface {
 		Get(
 			namespaceID string,
 			taskQueue string,
 			taskQueueType enumspb.TaskQueueType,
 			deploymentName string,
 			buildID string,
-		) (isMember bool, ok bool)
+		) (isMember bool, shouldSkipReactivation bool, revisionNumber int64, ok bool)
 
 		Put(
 			namespaceID string,
@@ -29,6 +36,8 @@ type (
 			deploymentName string,
 			buildID string,
 			isMember bool,
+			shouldSkipReactivation bool,
+			revisionNumber int64,
 		)
 	}
 
@@ -40,28 +49,34 @@ type (
 		buildID        string
 	}
 
-	VersionMembershipCacheImpl struct {
+	versionTaskQueueInfoCacheValue struct {
+		isMember               bool
+		shouldSkipReactivation bool  // false = unknown / not-found / eligible-for-reactivation
+		revisionNumber         int64 // 0 = unknown (old matching server or legacy format)
+	}
+
+	VersionMembershipAndReactivationStatusCacheImpl struct {
 		cache.Cache
 		metricsHandler metrics.Handler
 	}
 )
 
-// NewVersionMembershipCache wraps the provided cache with a typed API and metrics.
-func NewVersionMembershipCache(c cache.Cache, metricsHandler metrics.Handler) VersionMembershipCache {
+// NewVersionMembershipAndReactivationStatusCache wraps the provided cache with a typed API and metrics.
+func NewVersionMembershipAndReactivationStatusCache(c cache.Cache, metricsHandler metrics.Handler) VersionMembershipAndReactivationStatusCache {
 	h := metricsHandler.WithTags(metrics.CacheTypeTag(metrics.VersionMembershipCacheTypeTagValue))
-	return &VersionMembershipCacheImpl{
+	return &VersionMembershipAndReactivationStatusCacheImpl{
 		Cache:          c,
 		metricsHandler: h,
 	}
 }
 
-func (c *VersionMembershipCacheImpl) Get(
+func (c *VersionMembershipAndReactivationStatusCacheImpl) Get(
 	namespaceID string,
 	taskQueue string,
 	taskQueueType enumspb.TaskQueueType,
 	deploymentName string,
 	buildID string,
-) (isMember bool, ok bool) {
+) (isMember bool, shouldSkipReactivation bool, revisionNumber int64, ok bool) {
 	handler := c.metricsHandler.WithTags(metrics.OperationTag(metrics.VersionMembershipCacheGetScope), metrics.NamespaceIDTag(namespaceID))
 	metrics.CacheRequests.With(handler).Record(1)
 
@@ -75,24 +90,26 @@ func (c *VersionMembershipCacheImpl) Get(
 	v := c.Cache.Get(key)
 	if v == nil {
 		metrics.CacheMissCounter.With(handler).Record(1)
-		return false, false
+		return false, false, 0, false
 	}
-	isMember, ok = v.(bool)
+	value, ok := v.(versionTaskQueueInfoCacheValue)
 	if !ok {
 		// Unexpected type: treat as miss to avoid false positives.
 		metrics.CacheMissCounter.With(handler).Record(1)
-		return false, false
+		return false, false, 0, false
 	}
-	return isMember, true
+	return value.isMember, value.shouldSkipReactivation, value.revisionNumber, true
 }
 
-func (c *VersionMembershipCacheImpl) Put(
+func (c *VersionMembershipAndReactivationStatusCacheImpl) Put(
 	namespaceID string,
 	taskQueue string,
 	taskQueueType enumspb.TaskQueueType,
 	deploymentName string,
 	buildID string,
 	isMember bool,
+	shouldSkipReactivation bool,
+	revisionNumber int64,
 ) {
 	handler := c.metricsHandler.WithTags(metrics.OperationTag(metrics.VersionMembershipCachePutScope), metrics.NamespaceIDTag(namespaceID))
 	metrics.CacheRequests.With(handler).Record(1)
@@ -104,5 +121,9 @@ func (c *VersionMembershipCacheImpl) Put(
 		deploymentName: deploymentName,
 		buildID:        buildID,
 	}
-	c.Cache.Put(key, isMember)
+	c.Cache.Put(key, versionTaskQueueInfoCacheValue{
+		isMember:               isMember,
+		shouldSkipReactivation: shouldSkipReactivation,
+		revisionNumber:         revisionNumber,
+	})
 }